Repository: streamlink/streamlink

Issue:
session: don't load all built-in plugins at once
I've talked about the problem of how Streamlink loads its plugins multiple times, eg. [last year in the proposal for the new plugin matchers](https://github.com/streamlink/streamlink/issues/3814#issuecomment-871722799), but I never opened a dedicated thread for this, so let's do this. Don't expect anything from this for now though, I just want to write this down.

----

### Issue

Every time a new Streamlink session gets initialized, it always loads all built-in plugins at once, so that it can read the URL regexes of the plugin classes for its URL-to-plugin resolver and so that plugin arguments become available in streamlink_cli.

The big issue with that is, despite plugin files being byte-compiled by the python interpreter for faster loading times, that there's a lot of runtime logic in the plugin modules which unnecessarily delays the session initialization and which stores a ton of unneeded junk in memory as plugin references are kept by the session's `plugins` dict.

Loading only the plugin that's needed to find streams from the input URL would mean no wasted time during session initialization, no wasted memory, and lots of module imports like certain stream implementations for example could also be avoided. On fast systems, this is all negligible, but why be wasteful and have a plugin loading system like this?

I would even say that resolving this plugin loading issue could lead to some of the plugin request/merging rules being loosened, as adding more built-in plugins wouldn't be too bad. But that's a different topic.

### Solution

The idea of the added plugin matchers API last year wasn't just to allow defining URLs in a declarative way, but also making it easier for a parser to analyze the AST of plugin files, so that a pre-build step when installing or building wheels could generate a JSON file which could then be loaded by the session as an alternative to the plugin modules themselves.

I've recently had another go at it and got the bare logic working, so I'm pretty much aware of what's needed to be done in order to properly implement this.

In addition to the plugin matchers, I would prefer having plugin arguments defined as class decorators as well, so that the parser can be simplified. I've already written a parser for the documentation's plugin matrix which is based on the current plugin arguments definitions, but that can be changed easily. Moving the plugin args is simple and won't be a breaking change for custom plugins.

### Plan

1. Turn plugin arguments to class decorators, so that both URL regexes and arguments can be defined in the same declarative style
2. Add a pre-build step when installing / building wheels
   - Implement plugin AST visitor which outputs data containing a list of URL regexes+priorities and plugin arguments
   - Add a `build_py` setuptools hook (via versioningit's cmdclass wrapper) which writes the data as JSON
   - This should not affect sdists and editable installs
3. Rewrite `Session` plugin loading logic
   - Load JSON file if it exists, fall back to load-everything logic
   - Resolve from JSON if available, only load resolved plugin
   - Always load all custom plugins and resolve accordingly
4. Rewrite `streamlink_cli.main.setup_plugin_args()`
   - All available plugin arguments need to be loaded from JSON because of config files containing those args
   - The `parser_helper` function used when building the docs should always load all built-in plugins
5. Optionally add CLI arg for loading all plugins

Discussion:
gravyboat: Reviewed this last night but had to put some thought in to your points:

> The plugins data loading implementation always compares the checksum of the _plugins.json file (stored in the package RECORDS metadata) before evaluating it, so users can't easily modify it. I think this is better than applying a validation schema to the JSON data and unnecessarily slowing down the session initialization.

I'm fine with this.

> What I don't like though is that the match_url() debug log message when loading the resolved plugin is written to the output before the environment/package/config debug messages. This is because of the CLI implementation which wants to load all plugin config files first, which is only possible after resolving the plugin from the input URL.

Is this really that big a deal? I know it is a bit annoying but are there cases where we'll be somehow missing debug data that is written later or that will result in more difficult debugging because of this?

> Alternatively, the lazy plugins loading could've been implemented using Python's [pickle](https://docs.python.org/3/library/pickle.html) module, which would probably lead to better loading times, but this would come at the cost of not being able to read the plugins data, unlike JSON.

Let's avoid pickle, load times are good enough and being able to read the plugin data is useful.

I won't mark as approved in case there is anything else you want the check, but the changes seem fine to me.

Patch:
diff --git a/src/streamlink/session/plugins.py b/src/streamlink/session/plugins.py
index 554627a9..7842a17e 100644
--- a/src/streamlink/session/plugins.py
+++ b/src/streamlink/session/plugins.py
@@ -1,16 +1,28 @@
+import base64
+import hashlib
+import importlib.metadata
+import json
 import logging
 import pkgutil
+import re
+from contextlib import suppress
 from importlib.abc import PathEntryFinder
 from pathlib import Path
 from types import ModuleType
-from typing import Dict, Iterator, List, Optional, Tuple, Type, Union
+from typing import Dict, Iterator, List, Literal, Optional, Tuple, Type, Union
 
 import streamlink.plugins
-from streamlink.options import Arguments
+from streamlink.options import Argument, Arguments
 
 # noinspection PyProtectedMember
-from streamlink.plugin.plugin import NO_PRIORITY, Matchers, Plugin
-from streamlink.utils.module import exec_module
+from streamlink.plugin.plugin import _PLUGINARGUMENT_TYPE_REGISTRY, NO_PRIORITY, NORMAL_PRIORITY, Matcher, Matchers, Plugin
+from streamlink.utils.module import exec_module, get_finder
+
+
+try:
+    from typing import TypeAlias, TypedDict  # type: ignore[attr-defined]
+except ImportError:  # pragma: no cover
+    from typing_extensions import TypeAlias, TypedDict
 
 
 log = logging.getLogger(".".join(__name__.split(".")[:-1]))
@@ -18,18 +30,47 @@ log = logging.getLogger(".".join(__name__.split(".")[:-1]))
 # The path to Streamlink's built-in plugins
 _PLUGINS_PATH = Path(streamlink.plugins.__path__[0])
 
+# Hardcoded plugins JSON file path
+_PLUGINSDATA_PATH = _PLUGINS_PATH / "_plugins.json"
+# Hardcoded package name to look up metadata
+_PLUGINSDATA_PACKAGENAME = "streamlink"
+# The `parts` value of the plugins JSON file contained in the package's `RECORD` metadata file
+_PLUGINSDATA_PACKAGEPATH = "streamlink", "plugins", "_plugins.json"
+
 
 class StreamlinkPlugins:
     """
     Streamlink's session-plugins implementation. This class is responsible for loading plugins and resolving them from URLs.
 
     See the :attr:`Streamlink.plugins <streamlink.session.Streamlink.plugins>` attribute.
+
+    Unless disabled by the user, Streamlink will try to load built-in plugins lazily, when accessing them for the first time
+    while resolving input URLs. This is done by reading and interpreting serialized data of each plugin's
+    :func:`pluginmatcher <streamlink.plugin.pluginmatcher>` and :func:`pluginargument <streamlink.plugin.pluginargument>`
+    data from a pre-built plugins JSON file which is included in Streamlink's wheel packages.
+
+    Plugins which are sideloaded, either from specific user directories or custom input directories,
+    always have a higher priority than built-in plugins.
     """
 
-    def __init__(self, builtin: bool = True):
+    def __init__(self, builtin: bool = True, lazy: bool = True):
+        # Loaded plugins
         self._plugins: Dict[str, Type[Plugin]] = {}
 
-        if builtin:
+        # Data of built-in plugins which can be loaded lazily
+        self._matchers: Dict[str, Matchers] = {}
+        self._arguments: Dict[str, Arguments] = {}
+
+        # Attempt to load built-in plugins lazily first
+        if builtin and lazy:
+            data = StreamlinkPluginsData.load()
+            if data:
+                self._matchers, self._arguments = data
+            else:
+                lazy = False
+
+        # Load built-ins if lazy-loading is disabled or if loading plugins data has failed
+        if builtin and not lazy:
             self.load_builtin()
 
     def __getitem__(self, item: str) -> Type[Plugin]:
@@ -49,8 +90,8 @@ class StreamlinkPlugins:
         return item in self._plugins
 
     def get_names(self) -> List[str]:
-        """Get a list of the names of all loaded plugins"""
-        return sorted(self._plugins.keys())
+        """Get a list of the names of all available plugins"""
+        return sorted(self._plugins.keys() | self._matchers.keys())
 
     def get_loaded(self) -> Dict[str, Type[Plugin]]:
         """Get a mapping of all loaded plugins"""
@@ -62,7 +103,7 @@ class StreamlinkPlugins:
 
     def load_path(self, path: Union[Path, str]) -> bool:
         """Load plugins from a custom directory"""
-        plugins = self._load_from_path(path)
+        plugins = self._load_plugins_from_path(path)
         self.update(plugins)
 
         return bool(plugins)
@@ -82,6 +123,11 @@ class StreamlinkPlugins:
             for name, plugin in self._plugins.items()
             if plugin.arguments
         )
+        yield from (
+            (name, arguments)
+            for name, arguments in self._arguments.items()
+            if arguments and name not in self._plugins
+        )
 
     def iter_matchers(self) -> Iterator[Tuple[str, Matchers]]:
         """Iterate through all plugins and their :class:`Matchers <streamlink.plugin.plugin.Matchers>`"""
@@ -90,9 +136,14 @@ class StreamlinkPlugins:
             for name, plugin in self._plugins.items()
             if plugin.matchers
         )
+        yield from (
+            (name, matchers)
+            for name, matchers in self._matchers.items()
+            if matchers and name not in self._plugins
+        )
 
     def match_url(self, url: str) -> Optional[Tuple[str, Type[Plugin]]]:
-        """Find a matching plugin by URL"""
+        """Find a matching plugin by URL and load plugins which haven't been loaded yet"""
         match: Optional[str] = None
         priority: int = NO_PRIORITY
 
@@ -105,9 +156,23 @@ class StreamlinkPlugins:
         if match is None:
             return None
 
+        # plugin not loaded yet?
+        # if a custom plugin with the same name has already been loaded, skip loading the built-in plugin
+        if match not in self._plugins:
+            log.debug(f"Loading plugin: {match}")
+            lookup = self._load_plugin_from_path(match, _PLUGINS_PATH)
+            if not lookup:
+                return None
+            self._plugins[match] = lookup[1]
+
         return match, self._plugins[match]
 
-    def _load_from_path(self, path: Union[Path, str]) -> Dict[str, Type[Plugin]]:
+    def _load_plugin_from_path(self, name: str, path: Path) -> Optional[Tuple[ModuleType, Type[Plugin]]]:
+        finder = get_finder(path)
+
+        return self._load_plugin_from_finder(name, finder)
+
+    def _load_plugins_from_path(self, path: Union[Path, str]) -> Dict[str, Type[Plugin]]:
         plugins: Dict[str, Type[Plugin]] = {}
         for finder, name, _ in pkgutil.iter_modules([str(path)]):
             lookup = self._load_plugin_from_finder(name, finder=finder)  # type: ignore[arg-type]
@@ -133,3 +198,161 @@ class StreamlinkPlugins:
             return None
 
         return mod, mod.__plugin__
+
+
+_TListOfConstants: TypeAlias = List[Union[None, bool, int, float, str]]
+_TConstantOrListOfConstants: TypeAlias = Union[None, bool, int, float, str, _TListOfConstants]
+_TMappingOfConstantOrListOfConstants: TypeAlias = Dict[str, _TConstantOrListOfConstants]
+
+
+class _TPluginMatcherData(TypedDict):
+    pattern: str
+    flags: Optional[int]
+    priority: Optional[int]
+    name: Optional[str]
+
+
+class _TPluginArgumentData(TypedDict):
+    name: str
+    action: Optional[str]
+    nargs: Optional[Union[int, Literal["*", "?", "+"]]]
+    const: _TConstantOrListOfConstants
+    default: _TConstantOrListOfConstants
+    type: Optional[str]
+    type_args: Optional[_TListOfConstants]
+    type_kwargs: Optional[_TMappingOfConstantOrListOfConstants]
+    choices: Optional[_TListOfConstants]
+    required: Optional[bool]
+    help: Optional[str]
+    metavar: Optional[Union[str, List[str]]]
+    dest: Optional[str]
+    requires: Optional[Union[str, List[str]]]
+    prompt: Optional[str]
+    sensitive: Optional[bool]
+    argument_name: Optional[str]
+
+
+class _TPluginData(TypedDict):
+    matchers: List[_TPluginMatcherData]
+    arguments: List[_TPluginArgumentData]
+
+
+class StreamlinkPluginsData:
+    @classmethod
+    def load(cls) -> Optional[Tuple[Dict[str, Matchers], Dict[str, Arguments]]]:
+        # specific errors get logged, others are ignored intentionally
+        with suppress(Exception):
+            content = _PLUGINSDATA_PATH.read_bytes()
+
+            cls._validate(content)
+
+            return cls._parse(content)
+
+        return None
+
+    @staticmethod
+    def _validate(content: bytes) -> None:
+        # find plugins JSON checksum in package metadata
+        # https://packaging.python.org/en/latest/specifications/recording-installed-packages/#the-record-file
+        mode, filehash = next(
+            (packagepath.hash.mode, packagepath.hash.value)
+            for packagepath in importlib.metadata.files(_PLUGINSDATA_PACKAGENAME) or []
+            if packagepath.hash is not None and packagepath.parts == _PLUGINSDATA_PACKAGEPATH
+        )
+        if mode not in hashlib.algorithms_guaranteed or not filehash:
+            log.error("Unknown plugins data hash mode, falling back to loading all plugins")
+            raise Exception
+
+        # compare checksums
+        hashalg = getattr(hashlib, mode)
+        hashobj = hashalg(content)
+        digest = base64.urlsafe_b64encode(hashobj.digest()).decode("utf-8").rstrip("=")
+        if digest != filehash:
+            log.error("Plugins data checksum mismatch, falling back to loading all plugins")
+            raise Exception
+
+    @classmethod
+    def _parse(cls, content: bytes) -> Tuple[Dict[str, Matchers], Dict[str, Arguments]]:
+        data: Dict[str, _TPluginData] = json.loads(content)
+
+        try:
+            matchers = cls._build_matchers(data)
+        except Exception:
+            log.exception("Error while loading pluginmatcher data from JSON")
+            raise
+
+        try:
+            arguments = cls._build_arguments(data)
+        except Exception:
+            log.exception("Error while loading pluginargument data from JSON")
+            raise
+
+        return matchers, arguments
+
+    @classmethod
+    def _build_matchers(cls, data: Dict[str, _TPluginData]) -> Dict[str, Matchers]:
+        res = {}
+        for pluginname, plugindata in data.items():
+            matchers = Matchers()
+            for m in plugindata.get("matchers") or []:
+                matcher = cls._build_matcher(m)
+                matchers.register(matcher)
+
+            res[pluginname] = matchers
+
+        return res
+
+    @staticmethod
+    def _build_matcher(data: _TPluginMatcherData) -> Matcher:
+        return Matcher(
+            pattern=re.compile(data.get("pattern"), data.get("flags") or 0),
+            priority=data.get("priority") or NORMAL_PRIORITY,
+            name=data.get("name"),
+        )
+
+    @classmethod
+    def _build_arguments(cls, data: Dict[str, _TPluginData]) -> Dict[str, Arguments]:
+        res = {}
+        for pluginname, plugindata in data.items():
+            if not plugindata.get("arguments"):
+                continue
+            arguments = Arguments()
+            for a in plugindata.get("arguments") or []:
+                if argument := cls._build_argument(a):
+                    arguments.add(argument)
+
+            res[pluginname] = arguments
+
+        return res
+
+    @staticmethod
+    def _build_argument(data: _TPluginArgumentData) -> Optional[Argument]:
+        name: str = data.get("name")  # type: ignore[assignment]
+        _typedata = data.get("type")
+        if not _typedata:
+            _type = None
+        elif _type := _PLUGINARGUMENT_TYPE_REGISTRY.get(_typedata):
+            _type_args = data.get("type_args") or ()
+            _type_kwargs = data.get("type_kwargs") or {}
+            if _type_args or _type_kwargs:
+                _type = _type(*_type_args, **_type_kwargs)
+        else:
+            return None
+
+        return Argument(
+            name=name,
+            action=data.get("action"),
+            nargs=data.get("nargs"),
+            const=data.get("const"),
+            default=data.get("default"),
+            type=_type,
+            choices=data.get("choices"),
+            required=data.get("required") or False,
+            help=data.get("help"),
+            metavar=data.get("metavar"),
+            dest=data.get("dest"),
+            requires=data.get("requires"),
+            prompt=data.get("prompt"),
+            sensitive=data.get("sensitive") or False,
+            argument_name=data.get("argument_name"),
+        )
diff --git a/src/streamlink/session/session.py b/src/streamlink/session/session.py
index 5b84ece6..ea567f01 100644
--- a/src/streamlink/session/session.py
+++ b/src/streamlink/session/session.py
@@ -30,10 +30,13 @@ class Streamlink:
         options: Optional[Dict[str, Any]] = None,
         *,
         plugins_builtin: bool = True,
+        plugins_lazy: bool = True,
     ):
         """
         :param options: Custom options
         :param plugins_builtin: Whether to load built-in plugins or not
+        :param plugins_lazy: Load built-in plugins lazily. This option falls back to loading all built-in plugins
+                             if the pre-built plugin JSON metadata is not available (e.g. in editable installs) or is invalid.
         """
 
         #: An instance of Streamlink's :class:`requests.Session` subclass.
@@ -48,7 +51,7 @@ class Streamlink:
             self.options.update(options)
 
         #: Plugins of this session instance.
-        self.plugins: StreamlinkPlugins = StreamlinkPlugins(builtin=plugins_builtin)
+        self.plugins: StreamlinkPlugins = StreamlinkPlugins(builtin=plugins_builtin, lazy=plugins_lazy)
 
     def set_option(self, key: str, value: Any) -> None:
         """
diff --git a/src/streamlink/utils/module.py b/src/streamlink/utils/module.py
index 850c58c8..ca0b48f8 100644
--- a/src/streamlink/utils/module.py
+++ b/src/streamlink/utils/module.py
@@ -7,12 +7,18 @@ from types import ModuleType
 from typing import Union
 
 
-def load_module(name: str, path: Union[Path, str]) -> ModuleType:
+def get_finder(path: Union[Path, str]) -> PathEntryFinder:
     path = str(path)
     finder = get_importer(path)
     if not finder:
         raise ImportError(f"Not a package path: {path}", path=path)
 
+    return finder
+
+
+def load_module(name: str, path: Union[Path, str]) -> ModuleType:
+    finder = get_finder(path)
+
     return exec_module(finder, name)
 
 
diff --git a/src/streamlink_cli/_parser.py b/src/streamlink_cli/_parser.py
index 8b3efea1..ba6fac41 100644
--- a/src/streamlink_cli/_parser.py
+++ b/src/streamlink_cli/_parser.py
@@ -9,7 +9,7 @@ from streamlink_cli.main import setup_plugin_args
 
 
 def get_parser():
-    session = Streamlink(plugins_builtin=True)
+    session = Streamlink(plugins_builtin=True, plugins_lazy=False)
     parser = build_parser()
     setup_plugin_args(session, parser)
 
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index fc2a94ed..c637b63e 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -607,7 +607,7 @@ def print_plugins():
     if args.json:
         console.msg_json(pluginlist)
     else:
-        console.msg(f"Loaded plugins: {', '.join(pluginlist)}")
+        console.msg(f"Available plugins: {', '.join(pluginlist)}")
 
 
 def load_plugins(dirs: List[Path], showwarning: bool = True):


Test patch:
diff --git a/tests/cli/test_main.py b/tests/cli/test_main.py
index 11b8c49a..caaa5836 100644
--- a/tests/cli/test_main.py
+++ b/tests/cli/test_main.py
@@ -889,7 +889,7 @@ class TestCLIMainPrint(unittest.TestCase):
     @patch("sys.argv", ["streamlink", "--plugins"])
     def test_print_plugins(self, mock_stdout):
         self.subject()
-        assert self.get_stdout(mock_stdout) == "Loaded plugins: testplugin\n"
+        assert self.get_stdout(mock_stdout) == "Available plugins: testplugin\n"
 
     @patch("sys.stdout")
     @patch("sys.argv", ["streamlink", "--plugins", "--json"])
diff --git a/tests/conftest.py b/tests/conftest.py
index df1cb252..58fa8630 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -70,10 +70,12 @@ def _check_test_condition(item: pytest.Item):  # pragma: no cover
 def session(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch):
     options = getattr(request, "param", {})
     plugins_builtin = options.pop("plugins-builtin", False)
+    plugins_lazy = options.pop("plugins-lazy", False)
 
     session = Streamlink(
         options=options,
         plugins_builtin=plugins_builtin,
+        plugins_lazy=plugins_lazy,
     )
 
     try:
diff --git a/tests/session/test_plugins.py b/tests/session/test_plugins.py
index e713f2c8..acde293c 100644
--- a/tests/session/test_plugins.py
+++ b/tests/session/test_plugins.py
@@ -1,14 +1,35 @@
+import base64
+import hashlib
+import re
+
+# noinspection PyProtectedMember
+from importlib.metadata import FileHash, PackagePath
 from pathlib import Path
-from typing import Type
+from typing import Optional, Type, cast
 from unittest.mock import Mock, call
 
 import pytest
 
 import streamlink.plugins
 import tests.plugin
-from streamlink.plugin.plugin import Plugin, pluginargument
+from streamlink.options import Argument, Arguments
+
+# noinspection PyProtectedMember
+from streamlink.plugin.plugin import (
+    HIGH_PRIORITY,
+    LOW_PRIORITY,
+    NO_PRIORITY,
+    NORMAL_PRIORITY,
+    Matcher,
+    Matchers,
+    Plugin,
+    pluginargument,
+    pluginmatcher,
+)
 from streamlink.session import Streamlink
 from streamlink.session.plugins import StreamlinkPlugins
+from streamlink.utils.args import boolean, comma_list_filter
+from tests.plugin.testplugin import TestPlugin as _TestPlugin
 
 
 PATH_BUILTINPLUGINS = Path(streamlink.plugins.__path__[0])
@@ -16,6 +37,10 @@ PATH_TESTPLUGINS = Path(tests.plugin.__path__[0])
 PATH_TESTPLUGINS_OVERRIDE = PATH_TESTPLUGINS / "override"
 
 
+class _Plugin(Plugin):
+    def _get_streams(self): pass  # pragma: no cover
+
+
 @pytest.fixture(autouse=True)
 def caplog(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
     caplog.set_level(1, "streamlink")
@@ -24,13 +49,12 @@ def caplog(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
 
 @pytest.fixture(scope="module")
 def fake_plugin():
+    @pluginmatcher(re.compile("fake"))
     @pluginargument("foo")
     @pluginargument("bar")
-    class FakePlugin(Plugin):
+    class FakePlugin(_Plugin):
         __module__ = "streamlink.plugins.fake"
 
-        def _get_streams(self): pass  # pragma: no cover
-
     return FakePlugin
 
 
@@ -81,8 +105,8 @@ def test_iter_arguments(session: Streamlink, fake_plugin: Type[Plugin]):
 class TestLoad:
     def test_load_builtin(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, fake_plugin: Type[Plugin]):
         mock = Mock(return_value={"fake": fake_plugin})
-        monkeypatch.setattr(StreamlinkPlugins, "_load_from_path", mock)
-        session = Streamlink(plugins_builtin=True)
+        monkeypatch.setattr(StreamlinkPlugins, "_load_plugins_from_path", mock)
+        session = Streamlink(plugins_builtin=True, plugins_lazy=False)
 
         assert mock.call_args_list == [call(PATH_BUILTINPLUGINS)]
         assert "fake" in session.plugins
@@ -152,3 +176,369 @@ class TestLoad:
             session.plugins.load_path(PATH_TESTPLUGINS)
         assert session.plugins.get_names() == []
         assert caplog.record_tuples == []
+
+
+class TestLoadPluginsData:
+    @pytest.fixture()
+    def session(self, monkeypatch: pytest.MonkeyPatch, fake_plugin: Type[Plugin], metadata_files: Mock):
+        class MockStreamlinkPlugins(StreamlinkPlugins):
+            def load_builtin(self):
+                self._plugins.update({"fake": fake_plugin})
+                return True
+
+        monkeypatch.setattr("streamlink.session.session.StreamlinkPlugins", MockStreamlinkPlugins)
+
+        return Streamlink(plugins_builtin=True, plugins_lazy=True)
+
+    @pytest.fixture(autouse=True)
+    def metadata_files(self, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch, pluginsdata: Optional[str]):
+        options = getattr(request, "param", {})
+        package_record = options.get("package-record", True)
+        mode = options.get("package-record-hash-mode", "sha256")
+        filehash = options.get("package-record-hash-value", None)
+
+        mock = Mock(return_value=[])
+        monkeypatch.setattr("importlib.metadata.files", mock)
+
+        if not package_record or pluginsdata is None:
+            yield mock
+            return
+
+        if filehash is None:
+            # https://packaging.python.org/en/latest/specifications/recording-installed-packages/#the-record-file
+            digest = hashlib.sha256(pluginsdata.encode("utf-8")).digest()
+            filehash = base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=")
+
+        packagepath = PackagePath("streamlink", "plugins", "_plugins.json")
+        packagepath.hash = FileHash(f"{mode}={filehash}")
+
+        mock.return_value = [packagepath]
+        yield mock
+
+        assert mock.call_args_list == [call("streamlink")]
+
+    @pytest.fixture()
+    def pluginsdata(self, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch):
+        data = getattr(request, "param", "{}")
+
+        mock = Mock()
+        monkeypatch.setattr("streamlink.session.plugins._PLUGINSDATA_PATH", mock)
+
+        if data is None:
+            mock.read_bytes.side_effect = FileNotFoundError
+        else:
+            mock.read_bytes.return_value = data.encode("utf-8")
+
+        return data
+
+    # noinspection PyTestParametrized
+    @pytest.mark.parametrize(("metadata_files", "pluginsdata", "logs"), [
+        pytest.param(
+            {},
+            None,
+            [],
+            id="empty-json-file",
+        ),
+        pytest.param(
+            {"package-record": False},
+            """{}""",
+            [],
+            id="no-package-record",
+        ),
+        pytest.param(
+            {"package-record-hash-mode": "unknown"},
+            """{}""",
+            [("streamlink.session", "error", "Unknown plugins data hash mode, falling back to loading all plugins")],
+            id="invalid-package-record-hash-mode",
+        ),
+        pytest.param(
+            {"package-record-hash-value": "invalid"},
+            """{}""",
+            [("streamlink.session", "error", "Plugins data checksum mismatch, falling back to loading all plugins")],
+            id="invalid-package-record-hash-value",
+        ),
+    ], indirect=["metadata_files", "pluginsdata"])
+    def test_fallback_load_builtin(self, caplog: pytest.LogCaptureFixture, session: Streamlink, logs: list):
+        assert session.plugins.get_names() == ["fake"]
+        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == logs
+
+    @pytest.mark.parametrize("pluginsdata", [
+        pytest.param(
+            # language=json
+            """
+                {
+                    "testpluginA": {
+                        "matchers": [
+                            {"pattern": "foo"},
+                            {"pattern": "bar", "flags": 64, "priority": 10, "name": "bar"}
+                        ],
+                        "arguments": []
+                    },
+                    "testpluginB": {
+                        "matchers": [
+                            {"pattern": "baz"}
+                        ],
+                        "arguments": []
+                    }
+                }
+            """,
+            id="matchers",
+        ),
+    ], indirect=True)
+    def test_matchers(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
+        assert "fake" not in session.plugins
+        assert "testpluginA" not in session.plugins
+        assert "testpluginB" not in session.plugins
+        assert session.plugins.get_names() == ["testpluginA", "testpluginB"]
+        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == []
+
+        matchers_a = Matchers()
+        matchers_a.register(Matcher(pattern=re.compile(r"foo"), priority=NORMAL_PRIORITY, name=None))
+        matchers_a.register(Matcher(pattern=re.compile(r"bar", re.VERBOSE), priority=10, name="bar"))
+        matchers_b = Matchers()
+        matchers_b.register(Matcher(pattern=re.compile(r"baz"), priority=NORMAL_PRIORITY, name=None))
+        assert list(session.plugins.iter_matchers()) == [("testpluginA", matchers_a), ("testpluginB", matchers_b)]
+
+    @pytest.mark.parametrize("pluginsdata", [
+        pytest.param(
+            # language=json
+            """
+                {
+                    "success": {
+                        "matchers": [{"pattern": "foo"}],
+                        "arguments": []
+                    },
+                    "fail": {
+                        "matchers": [{"pattern": {"invalid": "type"}}],
+                        "arguments": []
+                    }
+                }
+            """,
+            id="matchers",
+        ),
+    ], indirect=True)
+    def test_matchers_failure(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
+        assert "fake" in session.plugins
+        assert "success" not in session.plugins
+        assert "fail" not in session.plugins
+        assert session.plugins.get_names() == ["fake"]
+        assert [name for name, matchers in session.plugins.iter_matchers()] == ["fake"]
+        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == [
+            ("streamlink.session", "error", "Error while loading pluginmatcher data from JSON"),
+        ]
+
+    @pytest.mark.parametrize("pluginsdata", [
+        pytest.param(
+            # language=json
+            """
+                {
+                    "empty": {
+                        "matchers": [{"pattern": "foo"}]
+                    },
+                    "testpluginA": {
+                        "matchers": [{"pattern": "bar"}],
+                        "arguments": [
+                            {
+                                "name": "foo",
+                                "action": "store",
+                                "nargs": 1,
+                                "default": "foo",
+                                "choices": ["foo", "bar"],
+                                "required": true,
+                                "help": "foo",
+                                "metavar": "FOO",
+                                "dest": "oof",
+                                "argument_name": "oof"
+                            },
+                            {
+                                "name": "bar",
+                                "const": "bar"
+                            }
+                        ]
+                    },
+                    "testpluginB": {
+                        "matchers": [{"pattern": "baz"}],
+                        "arguments": [
+                            {
+                                "name": "invalid",
+                                "type": "type_which_does_not_exist"
+                            },
+                            {
+                                "name": "bool",
+                                "type": "bool"
+                            },
+                            {
+                                "name": "cmf",
+                                "type": "comma_list_filter",
+                                "type_args": [["1", "2", "3"]],
+                                "type_kwargs": {"unique": true}
+                            }
+                        ]
+                    }
+                }
+            """,
+            id="arguments",
+        ),
+    ], indirect=True)
+    def test_arguments(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
+        assert "fake" not in session.plugins
+        assert "testpluginA" not in session.plugins
+        assert "testpluginB" not in session.plugins
+        assert session.plugins.get_names() == ["empty", "testpluginA", "testpluginB"]
+        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == []
+
+        arguments_a = Arguments()
+        arguments_a.add(Argument(
+            name="foo",
+            action="store",
+            nargs=1,
+            default="foo",
+            choices=["foo", "bar"],
+            required=True,
+            help="foo",
+            metavar="FOO",
+            dest="oof",
+            argument_name="oof",
+        ))
+        arguments_a.add(Argument(
+            name="bar",
+            const="bar",
+        ))
+        arguments_b = Arguments()
+        arguments_b.add(Argument(
+            name="bool",
+            type=boolean,
+        ))
+        arguments_b.add(Argument(
+            name="cmf",
+            type=comma_list_filter(["1", "2", "3"], unique=True),
+        ))
+        assert list(session.plugins.iter_arguments()) == [("testpluginA", arguments_a), ("testpluginB", arguments_b)]
+
+    @pytest.mark.parametrize("pluginsdata", [
+        pytest.param(
+            # language=json
+            """
+                {
+                    "success": {
+                        "matchers": [{"pattern": "foo"}],
+                        "arguments": [{"name": "foo"}]
+                    },
+                    "fail": {
+                        "matchers": [{"pattern": "bar"}],
+                        "arguments": [{"name": {"invalid": "type"}}]
+                    }
+                }
+            """,
+            id="arguments",
+        ),
+    ], indirect=True)
+    def test_arguments_failure(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
+        assert "fake" in session.plugins
+        assert "success" not in session.plugins
+        assert "fail" not in session.plugins
+        assert session.plugins.get_names() == ["fake"]
+        assert [name for name, arguments in session.plugins.iter_arguments()] == ["fake"]
+        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == [
+            ("streamlink.session", "error", "Error while loading pluginargument data from JSON"),
+        ]
+
+
+class TestMatchURL:
+    def test_priority(self, session: Streamlink):
+        @pluginmatcher(priority=HIGH_PRIORITY, pattern=re.compile("^(high|normal|low|no)$"))
+        class HighPriority(_Plugin):
+            pass
+
+        @pluginmatcher(priority=NORMAL_PRIORITY, pattern=re.compile("^(normal|low|no)$"))
+        class NormalPriority(_Plugin):
+            pass
+
+        @pluginmatcher(priority=LOW_PRIORITY, pattern=re.compile("^(low|no)$"))
+        class LowPriority(_Plugin):
+            pass
+
+        @pluginmatcher(priority=NO_PRIORITY, pattern=re.compile("^no$"))
+        class NoPriority(_Plugin):
+            pass
+
+        session.plugins.update({
+            "high": HighPriority,
+            "normal": NormalPriority,
+            "low": LowPriority,
+            "no": NoPriority,
+        })
+
+        assert session.plugins.match_url("no") == ("high", HighPriority)
+        assert session.plugins.match_url("low") == ("high", HighPriority)
+        assert session.plugins.match_url("normal") == ("high", HighPriority)
+        assert session.plugins.match_url("high") == ("high", HighPriority)
+
+        del session.plugins["high"]
+
+        assert session.plugins.match_url("no") == ("normal", NormalPriority)
+        assert session.plugins.match_url("low") == ("normal", NormalPriority)
+        assert session.plugins.match_url("normal") == ("normal", NormalPriority)
+        assert session.plugins.match_url("high") is None
+
+    def test_no_priority(self, session: Streamlink):
+        @pluginmatcher(priority=NO_PRIORITY, pattern=re.compile("^no$"))
+        class NoPriority(_Plugin):
+            pass
+
+        session.plugins.update({
+            "no": NoPriority,
+        })
+        assert session.plugins.match_url("no") is None
+
+
+class TestMatchURLLoadLazy:
+    @pytest.fixture(autouse=True)
+    def _pluginsdir(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("streamlink.session.plugins._PLUGINS_PATH", PATH_TESTPLUGINS)
+
+    @pytest.fixture()
+    def _loaded_plugins(self, session: Streamlink):
+        # loaded built-in or custom plugin
+        session.plugins.update({"testplugin": _TestPlugin})
+        assert session.plugins.get_loaded() == {"testplugin": _TestPlugin}
+
+    @pytest.fixture()
+    def _loaded_matchers(self, session: Streamlink):
+        matchers = cast(Matchers, _TestPlugin.matchers)
+        session.plugins._matchers.update({"testplugin": matchers})
+        assert session.plugins.get_loaded() == {}
+        assert session.plugins.get_names() == ["testplugin"]
+
+    @pytest.mark.usefixtures("_loaded_plugins")
+    def test_loaded(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
+        assert session.plugins.match_url("http://test.se") == ("testplugin", _TestPlugin)
+        assert [(record.name, record.levelname, record.message) for record in caplog.records] == []
+
+    @pytest.mark.usefixtures("_loaded_matchers")
+    def test_load(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
+        lookup = session.plugins.match_url("http://test.se")
+        assert lookup is not None
+        name, plugin = lookup
+        assert name == "testplugin"
+        # can't compare plugin classes here due to exec_module(), so just compare module name, matchers and arguments
+        assert plugin.__module__ == "streamlink.plugins.testplugin"
+        assert plugin.matchers == _TestPlugin.matchers
+        assert plugin.arguments == _TestPlugin.arguments
+        assert "testplugin" in session.plugins.get_loaded()
+        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
+            ("streamlink.session", "debug", "Loading plugin: testplugin"),
+        ]
+
+    @pytest.mark.usefixtures("_loaded_matchers")
+    def test_fail_builtin(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink):
+        path = str(PATH_TESTPLUGINS / "testplugin.py")
+        mock = Mock(side_effect=ImportError("", path=path))
+        monkeypatch.setattr("streamlink.session.plugins.exec_module", mock)
+
+        assert session.plugins.match_url("http://test.se") is None
+        assert session.plugins.get_loaded() == {}
+        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
+            ("streamlink.session", "debug", "Loading plugin: testplugin"),
+            ("streamlink.session", "error", f"Failed to load plugin testplugin from {path}\n"),
+        ]
diff --git a/tests/session/test_session.py b/tests/session/test_session.py
index 54a9ccbd..8195102e 100644
--- a/tests/session/test_session.py
+++ b/tests/session/test_session.py
@@ -8,7 +8,7 @@ import requests_mock as rm
 import tests.plugin
 from streamlink.exceptions import NoPluginError, StreamlinkDeprecationWarning
 from streamlink.options import Options
-from streamlink.plugin import HIGH_PRIORITY, LOW_PRIORITY, NO_PRIORITY, NORMAL_PRIORITY, Plugin, pluginmatcher
+from streamlink.plugin import Plugin, pluginmatcher
 from streamlink.session import Streamlink
 from streamlink.stream.hls import HLSStream
 from streamlink.stream.http import HTTPStream
@@ -135,55 +135,6 @@ class TestResolveURL:
             session.resolve_url("http://secure")
         assert session.resolve_url("https://secure")[1] is PluginHttps
 
-    def test_resolve_url_priority(self, session: Streamlink):
-        @pluginmatcher(priority=HIGH_PRIORITY, pattern=re.compile(
-            "https://(high|normal|low|no)$",
-        ))
-        class HighPriority(_EmptyPlugin):
-            pass
-
-        @pluginmatcher(priority=NORMAL_PRIORITY, pattern=re.compile(
-            "https://(normal|low|no)$",
-        ))
-        class NormalPriority(_EmptyPlugin):
-            pass
-
-        @pluginmatcher(priority=LOW_PRIORITY, pattern=re.compile(
-            "https://(low|no)$",
-        ))
-        class LowPriority(_EmptyPlugin):
-            pass
-
-        @pluginmatcher(priority=NO_PRIORITY, pattern=re.compile(
-            "https://(no)$",
-        ))
-        class NoPriority(_EmptyPlugin):
-            pass
-
-        session.plugins.update({
-            "high": HighPriority,
-            "normal": NormalPriority,
-            "low": LowPriority,
-            "no": NoPriority,
-        })
-        no = session.resolve_url_no_redirect("no")[1]
-        low = session.resolve_url_no_redirect("low")[1]
-        normal = session.resolve_url_no_redirect("normal")[1]
-        high = session.resolve_url_no_redirect("high")[1]
-
-        assert no is HighPriority
-        assert low is HighPriority
-        assert normal is HighPriority
-        assert high is HighPriority
-
-        session.resolve_url.cache_clear()
-        session.plugins.clear()
-        session.plugins.update({
-            "no": NoPriority,
-        })
-        with pytest.raises(NoPluginError):
-            session.resolve_url_no_redirect("no")
-
 
 class TestStreams:
     @pytest.fixture(autouse=True)