Skip to content

isolate validator worker installs and raise default concurrency #719

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zouyonghe merged 4 commits into from
Apr 24, 2026
Merged

isolate validator worker installs and raise default concurrency #719

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c20f4e0
isolate validator worker installs and raise default concurrency
zouyonghe Apr 24, 2026
f603b2c
deduplicate validator worker install paths
zouyonghe Apr 24, 2026
f63c809
strengthen validator worker isolation test
zouyonghe Apr 24, 2026
05e874b
normalize validator worker path comparisons
zouyonghe Apr 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/validate-plugin-smoke.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ on:
max_workers:
description: "Maximum concurrent plugin validations"
required: false
default: "8"
default: "16"

jobs:
validate-plugin-smoke:
Expand All @@ -52,7 +52,7 @@ jobs:
echo "ASTRBOT_REF=master" >> "$GITHUB_ENV"
echo "PLUGIN_NAME_LIST=" >> "$GITHUB_ENV"
echo "PLUGIN_LIMIT=" >> "$GITHUB_ENV"
echo "MAX_WORKERS=8" >> "$GITHUB_ENV"
echo "MAX_WORKERS=16" >> "$GITHUB_ENV"
echo "SHOULD_VALIDATE=true" >> "$GITHUB_ENV"
echo "VALIDATION_NOTE=Running scheduled full plugin validation." >> "$GITHUB_ENV"

Expand Down
36 changes: 35 additions & 1 deletion scripts/validate_plugins/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@

REQUIRED_METADATA_FIELDS = ("name", "desc", "version", "author")
DEFAULT_CLONE_TIMEOUT = 120
DEFAULT_MAX_WORKERS = 8
DEFAULT_MAX_WORKERS = 16
CONFLICT_MARKERS = ("<<<<<<<", "=======", ">>>>>>>")


Expand Down Expand Up @@ -282,6 +282,38 @@ def build_worker_sys_path(*, astrbot_root: Path, astrbot_path: Path) -> list[str
return [str(astrbot_root.resolve()), str(astrbot_path.resolve())]


def normalize_path_for_comparison(path: str | os.PathLike[str]) -> str:
path_str = os.fspath(path)
return os.path.normcase(os.path.realpath(os.path.abspath(os.path.expanduser(path_str))))


def configure_worker_install_target(*, temp_root: Path) -> Path:
"""Configure process-global install/import state for a validator worker.

This mutates ``os.environ`` and ``sys.path`` for the lifetime of the worker
process so plugin dependency installs stay isolated under ``temp_root``.
"""

site_packages = (temp_root / "site-packages").resolve()
site_packages.mkdir(parents=True, exist_ok=True)
site_packages_str = str(site_packages)
site_packages_key = normalize_path_for_comparison(site_packages_str)

os.environ["PIP_TARGET"] = site_packages_str
existing_pythonpath = [
entry
for entry in os.environ.get("PYTHONPATH", "").split(os.pathsep)
if entry and normalize_path_for_comparison(entry) != site_packages_key
]
os.environ["PYTHONPATH"] = os.pathsep.join([site_packages_str, *existing_pythonpath])

sys.path[:] = [
entry for entry in sys.path if normalize_path_for_comparison(entry) != site_packages_key
]
sys.path.insert(0, site_packages_str)
return site_packages


def build_report(results: list[dict]) -> dict:
passed = sum(1 for result in results if result.get("severity") == "pass")
warned = sum(1 for result in results if result.get("severity") == "warn")
Expand Down Expand Up @@ -700,6 +732,8 @@ async def run_worker_load_check(plugin_dir_name: str, normalized_repo_url: str)
def run_worker(args: argparse.Namespace) -> int:
temp_root = Path(tempfile.mkdtemp(prefix="astrbot-plugin-worker-"))
try:
configure_worker_install_target(temp_root=temp_root)

astrbot_root = temp_root / "astrbot-root"
plugin_store = astrbot_root / "data" / "plugins"
plugin_config = astrbot_root / "data" / "config"
Expand Down
129 changes: 127 additions & 2 deletions tests/test_validate_plugins.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,12 +386,12 @@ def test_load_plugins_index_rejects_non_dict_values(self):


class ValidationProgressTests(unittest.TestCase):
def test_build_parser_defaults_max_workers_to_eight(self):
def test_build_parser_defaults_max_workers_to_sixteen(self):
module = load_validator_module()

args = module.build_parser().parse_args(["--astrbot-path", "/tmp/AstrBot"])

self.assertEqual(args.max_workers, 8)
self.assertEqual(args.max_workers, 16)

def test_build_parser_rejects_non_positive_worker_and_timeout_values(self):
module = load_validator_module()
Expand Down Expand Up @@ -748,6 +748,131 @@ async def load(self, specified_dir_name: str):
self.assertEqual(result["message"], "{'reason': 'boom'}")


class RunWorkerIsolationTests(unittest.TestCase):
def _assert_worker_isolated(self, *, temp_root: Path, args, observed: dict) -> None:
self.assertIn("astrbot_root", observed)
astrbot_root = Path(observed["astrbot_root"]).resolve()
shared_astrbot_path = Path(args.astrbot_path).resolve()

self.assertNotEqual(astrbot_root, shared_astrbot_path)
self.assertTrue(
astrbot_root.is_relative_to(temp_root.resolve()),
f"worker astrbot_root {astrbot_root} should be under {temp_root}",
)

current = astrbot_root
found_worker_prefix = False
while True:
if current.name.startswith("astrbot-plugin-worker-"):
found_worker_prefix = True
break
if current.parent == current:
break
current = current.parent

self.assertTrue(found_worker_prefix)
self.assertTrue((astrbot_root / "data" / "plugins").is_dir())
self.assertTrue((astrbot_root / "data" / "config").is_dir())

def test_configure_worker_install_target_deduplicates_process_paths(self):
module = load_validator_module()
original_sys_path = list(sys.path)

with tempfile.TemporaryDirectory() as tmp_dir:
temp_root = Path(tmp_dir)
site_packages = (temp_root / "site-packages").resolve()
site_packages_alias = os.path.join(str(site_packages.parent), ".", site_packages.name)
extra_path = (temp_root / "extra-path").resolve()
extra_path.mkdir()
Comment on lines +751 to +786
Copy link
Copy Markdown

@sourcery-ai sourcery-ai Bot Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): Strengthen run_worker isolation test by asserting properties of the temporary root, not just that it exists

The current test only checks that ASTRBOT_ROOT and PIP_TARGET are set and related, but not that the worker is actually isolated from the real AstrBot path. Consider also asserting that:

  • observed["astrbot_root"] differs from args.astrbot_path.
  • observed["astrbot_root"] is under a temporary worker directory (e.g., the astrbot-plugin-worker- prefix or a parent dir starting with it).
  • Optionally, the expected plugin store/config directories are created under this root, if that’s part of the isolation contract.

These checks would more directly validate that each worker uses its own temp root instead of the shared AstrBot path.

Suggested implementation:

class RunWorkerIsolationTests(unittest.TestCase):
    def _assert_worker_isolated(self, temp_root: Path, args, observed: dict) -> None:
        """
        Assert that a worker run is isolated from the real AstrBot path.

        This validates that:
        - The worker's astrbot_root differs from args.astrbot_path.
        - The astrbot_root resides under the temp_root used for the worker.
        - There exists a parent directory in the astrbot_root path whose name
          starts with the expected worker directory prefix.
        """
        self.assertIn("astrbot_root", observed, "observed must contain 'astrbot_root'")
        astrbot_root = Path(observed["astrbot_root"])

        # 1. The worker root must differ from the configured AstrBot path.
        self.assertNotEqual(
            astrbot_root,
            Path(args.astrbot_path),
            "Worker astrbot_root should not match the shared AstrBot path",
        )

        # 2. The worker root must be inside the temporary worker directory root.
        self.assertTrue(
            astrbot_root.is_relative_to(temp_root)
            if hasattr(astrbot_root, "is_relative_to")
            else str(astrbot_root).startswith(str(temp_root)),
            f"Worker astrbot_root ({astrbot_root}) is expected to be under temp_root ({temp_root})",
        )

        # 3. Somewhere in the ancestry we expect a worker directory prefix.
        worker_dir_prefix = "astrbot-plugin-worker-"
        current = astrbot_root
        found_prefix = False
        # Safeguard loop: walk up until root
        while True:
            if current.name.startswith(worker_dir_prefix):
                found_prefix = True
                break
            if current.parent == current:
                break
            current = current.parent

        self.assertTrue(
            found_prefix,
            f"No ancestor directory of astrbot_root ({astrbot_root}) "
            f"starts with '{worker_dir_prefix}'",
        )

    def test_configure_worker_install_target_deduplicates_process_paths(self):
        module = load_validator_module()
        original_sys_path = list(sys.path)

        with tempfile.TemporaryDirectory() as tmp_dir:
            temp_root = Path(tmp_dir)
            site_packages = (temp_root / "site-packages").resolve()
            extra_path = (temp_root / "extra-path").resolve()
            extra_path.mkdir()
            observed = {}

To fully implement the suggested stronger isolation checks, you should:

  1. Ensure that test_configure_worker_install_target_deduplicates_process_paths (or another test in RunWorkerIsolationTests) captures:

    • the parsed arguments (e.g., args = module.build_parser().parse_args([...])),
    • the temporary worker root (temp_root, already present),
    • the worker’s effective AstrBot root in observed["astrbot_root"] (this appears to already exist per your comment).

  2. After the worker has been configured/run and observed["astrbot_root"] is populated, call the helper:

    self._assert_worker_isolated(temp_root=temp_root, args=args, observed=observed)

  3. If part of the worker isolation contract is that specific plugin store/config directories are created under astrbot_root, extend _assert_worker_isolated with additional checks, for example:

    plugins_dir = astrbot_root / "plugins"
config_dir = astrbot_root / "config"
self.assertTrue(plugins_dir.is_dir())
self.assertTrue(config_dir.is_dir())

    and adjust the directory names to match your actual layout.

You’ll need to wire in step (2) at the appropriate point in the existing test body where observed is fully populated, since that portion of the test is not visible in the provided snippet.

observed = {}

sys.path[:0] = [str(site_packages), site_packages_alias]
with mock.patch.dict(
os.environ,
{"PYTHONPATH": os.pathsep.join([site_packages_alias, str(extra_path)])},
clear=True,
):
module.configure_worker_install_target(temp_root=temp_root)
module.configure_worker_install_target(temp_root=temp_root)

observed["pip_target"] = os.environ["PIP_TARGET"]
observed["pythonpath_entries"] = os.environ["PYTHONPATH"].split(os.pathsep)
observed["resolved_pythonpath_count"] = sum(
1
for entry in observed["pythonpath_entries"]
if Path(entry).resolve() == site_packages
)
observed["resolved_sys_path_count"] = sum(
1 for entry in sys.path if Path(entry).resolve() == site_packages
)

self.assertEqual(observed["pip_target"], str(site_packages))
self.assertEqual(observed["resolved_pythonpath_count"], 1)
self.assertEqual(observed["resolved_sys_path_count"], 1)
self.assertIn(str(extra_path), observed["pythonpath_entries"])

sys.path[:] = original_sys_path

def test_run_worker_isolates_plugin_installs_under_temp_root(self):
module = load_validator_module()
observed = {}
original_sys_path = list(sys.path)

async def fake_run_worker_load_check(plugin_dir_name: str, normalized_repo_url: str):
observed["plugin_dir_name"] = plugin_dir_name
observed["normalized_repo_url"] = normalized_repo_url
observed["astrbot_root"] = os.environ.get("ASTRBOT_ROOT")
observed["pip_target"] = os.environ.get("PIP_TARGET")
observed["sys_path"] = list(sys.path)
return module.build_result(
plugin=plugin_dir_name,
repo=normalized_repo_url,
normalized_repo_url=normalized_repo_url,
ok=True,
stage="load",
message="plugin loaded successfully",
plugin_dir_name=plugin_dir_name,
)

with tempfile.TemporaryDirectory() as tmp_dir:
worker_temp_root = Path(tmp_dir) / "astrbot-plugin-worker-test-root"
worker_temp_root.mkdir()
plugin_source_dir = Path(tmp_dir) / "plugin-src"
plugin_source_dir.mkdir()
(plugin_source_dir / "main.py").write_text("print('hello')\n", encoding="utf-8")
args = module.argparse.Namespace(
astrbot_path="/tmp/AstrBot",
plugin_source_dir=str(plugin_source_dir),
plugin_dir_name="demo_plugin",
normalized_repo_url="https://github.com/example/demo-plugin",
)

with mock.patch.dict(os.environ, {}, clear=True):
with mock.patch.object(module, "run_worker_load_check", side_effect=fake_run_worker_load_check):
with mock.patch.object(module.tempfile, "mkdtemp", return_value=str(worker_temp_root)):
with mock.patch.object(module.shutil, "rmtree") as rmtree_mock:
exit_code = module.run_worker(args)

self._assert_worker_isolated(temp_root=worker_temp_root, args=args, observed=observed)
rmtree_mock.assert_called_once_with(worker_temp_root, ignore_errors=True)

sys.path[:] = original_sys_path
Comment on lines +819 to +859
Copy link
Copy Markdown

@gemini-code-assist gemini-code-assist Bot Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The restoration of sys.path at the end of the test is not robust. If module.run_worker(args) or any assertion fails, the line sys.path[:] = original_sys_path will be skipped, leaving the global sys.path modified for subsequent tests in the same process. It is better to use a try...finally block or self.addCleanup to ensure the environment is restored regardless of the test outcome.

        original_sys_path = list(sys.path)
        def restore_path():
            sys.path[:] = original_sys_path
        self.addCleanup(restore_path)

        async def fake_run_worker_load_check(plugin_dir_name: str, normalized_repo_url: str):
            observed["plugin_dir_name"] = plugin_dir_name
            observed["normalized_repo_url"] = normalized_repo_url
            observed["astrbot_root"] = os.environ.get("ASTRBOT_ROOT")
            observed["pip_target"] = os.environ.get("PIP_TARGET")
            observed["sys_path"] = list(sys.path)
            return module.build_result(
                plugin=plugin_dir_name,
                repo=normalized_repo_url,
                normalized_repo_url=normalized_repo_url,
                ok=True,
                stage="load",
                message="plugin loaded successfully",
                plugin_dir_name=plugin_dir_name,
            )

        with tempfile.TemporaryDirectory() as tmp_dir:
            plugin_source_dir = Path(tmp_dir) / "plugin-src"
            plugin_source_dir.mkdir()
            (plugin_source_dir / "main.py").write_text("print('hello')\n", encoding="utf-8")
            args = module.argparse.Namespace(
                astrbot_path="/tmp/AstrBot",
                plugin_source_dir=str(plugin_source_dir),
                plugin_dir_name="demo_plugin",
                normalized_repo_url="https://github.com/example/demo-plugin",
            )

            with mock.patch.dict(os.environ, {}, clear=True):
                with mock.patch.object(module, "run_worker_load_check", side_effect=fake_run_worker_load_check):
                    exit_code = module.run_worker(args)


self.assertEqual(exit_code, 0)
self.assertEqual(observed["plugin_dir_name"], "demo_plugin")
self.assertEqual(
observed["normalized_repo_url"],
"https://github.com/example/demo-plugin",
)
self.assertIsNotNone(observed["astrbot_root"])
self.assertIsNotNone(observed["pip_target"])
self.assertEqual(
Path(observed["pip_target"]).resolve(),
(Path(observed["astrbot_root"]).parent / "site-packages").resolve(),
)
self.assertIn(observed["pip_target"], observed["sys_path"])


class ReportBuilderTests(unittest.TestCase):
def test_build_report_counts_passed_warned_and_failed_results(self):
module = load_validator_module()
Expand Down
Loading