849 lines
30 KiB
Python
Executable File
849 lines
30 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Resolve and execute Cargo commands in bounded, signature-stable build lanes.
|
|
|
|
This module intentionally uses only the Python standard library. It is also
|
|
imported by build_storage.py, so lane identity and sentinel validation have one
|
|
implementation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Mapping, Sequence
|
|
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError as error: # pragma: no cover - Python < 3.11
|
|
raise SystemExit("cargo_lane.py requires Python 3.11 or newer") from error
|
|
|
|
|
|
SCHEMA_VERSION = 1
|
|
SENTINEL_NAME = ".blacksite-cargo-lane.json"
|
|
LANE_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
|
|
SEPARATE_BUILD_DIR_MINIMUM = (1, 91, 0)
|
|
|
|
DEFAULT_BUILD_STORAGE: dict[str, Any] = {
|
|
"cache_root": "cargo-cache-home",
|
|
"workspace_partition": "canonical-path-hash",
|
|
"dev_target_dir": "target",
|
|
"exceptional_target_root": "target/lanes",
|
|
"persistent_lane": "dev",
|
|
"candidate_lane": "candidate",
|
|
"hot_reload_lane": "hot-reload",
|
|
"full_debug_lane": "full-debug",
|
|
"package_lane": "package",
|
|
}
|
|
|
|
|
|
class LaneError(RuntimeError):
|
|
"""A lane cannot be resolved or safely reused."""
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def canonical_path(path: Path | str) -> Path:
|
|
return Path(path).expanduser().resolve(strict=False)
|
|
|
|
|
|
def workspace_root(start: Path | None = None) -> Path:
|
|
override = os.environ.get("BLACKSITE_WORKSPACE_ROOT")
|
|
if override:
|
|
candidate = canonical_path(override)
|
|
elif start is not None:
|
|
candidate = canonical_path(start)
|
|
else:
|
|
candidate = canonical_path(Path(__file__).parents[2])
|
|
|
|
for current in (candidate, *candidate.parents):
|
|
if (current / "Cargo.toml").is_file() and (current / ".git").exists():
|
|
return current
|
|
raise LaneError(f"could not locate a Cargo workspace from {candidate}")
|
|
|
|
|
|
def workspace_hash(root: Path) -> str:
|
|
return hashlib.sha256(os.fsencode(str(canonical_path(root)))).hexdigest()[:20]
|
|
|
|
|
|
def cargo_home() -> Path:
|
|
return canonical_path(os.environ.get("CARGO_HOME", Path.home() / ".cargo"))
|
|
|
|
|
|
def load_workflow(root: Path) -> dict[str, Any]:
|
|
path = root / ".codex" / "workflow.toml"
|
|
document: dict[str, Any] = {}
|
|
if path.is_file():
|
|
with path.open("rb") as handle:
|
|
loaded = tomllib.load(handle)
|
|
if not isinstance(loaded, dict):
|
|
raise LaneError(f"workflow config is not a TOML table: {path}")
|
|
document = loaded
|
|
|
|
storage = dict(DEFAULT_BUILD_STORAGE)
|
|
configured = document.get("build_storage", {})
|
|
if configured:
|
|
if not isinstance(configured, dict):
|
|
raise LaneError("[build_storage] must be a TOML table")
|
|
storage.update(configured)
|
|
|
|
verification = document.get("verification", {})
|
|
if isinstance(verification, dict):
|
|
for source, destination in (
|
|
("persistent_lane", "persistent_lane"),
|
|
("candidate_lane", "candidate_lane"),
|
|
("hot_reload_lane", "hot_reload_lane"),
|
|
("full_debug_lane", "full_debug_lane"),
|
|
("package_lane", "package_lane"),
|
|
):
|
|
if source in verification:
|
|
storage[destination] = verification[source]
|
|
|
|
document["build_storage"] = storage
|
|
return document
|
|
|
|
|
|
def _configured_path(root: Path, value: str, *, base: Path | None = None) -> Path:
|
|
expanded = Path(os.path.expandvars(os.path.expanduser(value)))
|
|
if expanded.is_absolute():
|
|
return canonical_path(expanded)
|
|
return canonical_path((base or root) / expanded)
|
|
|
|
|
|
def cache_root(root: Path, workflow: Mapping[str, Any]) -> Path:
|
|
override = os.environ.get("BLACKSITE_BUILD_CACHE_ROOT")
|
|
if override:
|
|
return canonical_path(override)
|
|
|
|
storage = workflow["build_storage"]
|
|
configured = str(storage.get("cache_root", "cargo-cache-home"))
|
|
if configured == "cargo-cache-home":
|
|
return cargo_home() / "blacksite-build"
|
|
if configured == "workspace-parent-cache":
|
|
return canonical_path(root.parent / ".blacksite-build-cache")
|
|
return _configured_path(root, configured)
|
|
|
|
|
|
def parse_version(text: str) -> tuple[int, int, int]:
|
|
match = re.search(r"\b(\d+)\.(\d+)\.(\d+)", text)
|
|
if not match:
|
|
return (0, 0, 0)
|
|
return tuple(int(part) for part in match.groups()) # type: ignore[return-value]
|
|
|
|
|
|
def _capture(command: Sequence[str]) -> str:
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
except OSError as error:
|
|
return f"unavailable: {error}"
|
|
return completed.stdout.strip()
|
|
|
|
|
|
def toolchain_info() -> dict[str, Any]:
|
|
cargo_text = _capture(["cargo", "-Vv"])
|
|
rustc_text = _capture(["rustc", "-Vv"])
|
|
host = "unknown"
|
|
for line in rustc_text.splitlines():
|
|
if line.startswith("host:"):
|
|
host = line.partition(":")[2].strip()
|
|
break
|
|
version = parse_version(cargo_text)
|
|
return {
|
|
"cargo": cargo_text.splitlines()[0] if cargo_text else "unavailable",
|
|
"cargo_version": ".".join(str(part) for part in version),
|
|
"rustc": rustc_text.splitlines()[0] if rustc_text else "unavailable",
|
|
"host": host,
|
|
"separate_build_dir": version >= SEPARATE_BUILD_DIR_MINIMUM,
|
|
}
|
|
|
|
|
|
def _repo_target_config(root: Path, target: str) -> dict[str, Any]:
|
|
path = root / ".cargo" / "config.toml"
|
|
result: dict[str, Any] = {"linker": None, "rustflags": []}
|
|
if path.is_file():
|
|
with path.open("rb") as handle:
|
|
config = tomllib.load(handle)
|
|
target_table = config.get("target", {})
|
|
if isinstance(target_table, dict):
|
|
values = target_table.get(target, {})
|
|
if isinstance(values, dict):
|
|
result["linker"] = values.get("linker")
|
|
flags = values.get("rustflags", [])
|
|
if isinstance(flags, list):
|
|
result["rustflags"] = [str(flag) for flag in flags]
|
|
encoded_flags = os.environ.get("CARGO_ENCODED_RUSTFLAGS")
|
|
if encoded_flags:
|
|
result["rustflags"] = [flag for flag in encoded_flags.split("\x1f") if flag]
|
|
result["rustflags_source"] = "CARGO_ENCODED_RUSTFLAGS"
|
|
elif os.environ.get("RUSTFLAGS"):
|
|
result["rustflags"] = shlex.split(os.environ["RUSTFLAGS"])
|
|
result["rustflags_source"] = "environment"
|
|
else:
|
|
result["rustflags_source"] = str(path) if path.is_file() else "cargo-default"
|
|
result["rustc_wrapper"] = os.environ.get("RUSTC_WRAPPER")
|
|
linker_variable = "CARGO_TARGET_" + target.upper().replace("-", "_") + "_LINKER"
|
|
if os.environ.get(linker_variable):
|
|
result["linker"] = os.environ[linker_variable]
|
|
return result
|
|
|
|
|
|
def validate_lane_name(lane: str) -> str:
|
|
if not LANE_RE.fullmatch(lane):
|
|
raise LaneError(
|
|
f"invalid lane {lane!r}; use lowercase letters, digits, and hyphens"
|
|
)
|
|
return lane
|
|
|
|
|
|
def profile_directory(profile: str) -> str:
|
|
if profile in {"dev", "test"}:
|
|
return "debug"
|
|
if profile in {"release", "bench"}:
|
|
return "release"
|
|
return profile
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandContext:
|
|
profile: str
|
|
profile_dir: str
|
|
feature_signature: str
|
|
cargo_subcommand: str | None
|
|
target_triple: str | None
|
|
|
|
|
|
PROJECT_ALIAS_SUBCOMMANDS = {
|
|
"bake-navigation",
|
|
"package-project",
|
|
"process-assets",
|
|
"upgrade-project",
|
|
"validate-levels",
|
|
"validate-samples",
|
|
}
|
|
|
|
|
|
def command_context(command: Sequence[str], lane: str) -> CommandContext:
|
|
args = list(command)
|
|
cargo_index: int | None = None
|
|
for index, value in enumerate(args[:2]):
|
|
if Path(value).name == "cargo":
|
|
cargo_index = index
|
|
break
|
|
|
|
cargo_args = args[cargo_index + 1 :] if cargo_index is not None else []
|
|
subcommand = next((arg for arg in cargo_args if not arg.startswith("-")), None)
|
|
# Arguments following Blacksite's Cargo aliases are forwarded to the xtask
|
|
# binary by the alias's trailing `--`. They are not Cargo options. In
|
|
# particular, `cargo package-project --profile package-qa` selects the
|
|
# package manifest profile and must not repartition the Cargo build lane.
|
|
cargo_options = (
|
|
cargo_args[: cargo_args.index(subcommand) + 1]
|
|
if subcommand in PROJECT_ALIAS_SUBCOMMANDS
|
|
else cargo_args
|
|
)
|
|
if "--release" in cargo_options:
|
|
profile = "release"
|
|
else:
|
|
profile = "test" if subcommand == "test" else "dev"
|
|
for index, value in enumerate(cargo_options):
|
|
if value.startswith("--profile="):
|
|
profile = value.partition("=")[2]
|
|
elif value == "--profile" and index + 1 < len(cargo_options):
|
|
profile = cargo_options[index + 1]
|
|
|
|
features: set[str] = set()
|
|
all_features = "--all-features" in cargo_options
|
|
no_default = "--no-default-features" in cargo_options
|
|
for index, value in enumerate(cargo_options):
|
|
raw: str | None = None
|
|
if value.startswith("--features="):
|
|
raw = value.partition("=")[2]
|
|
elif value.startswith("-F") and value != "-F":
|
|
raw = value[2:]
|
|
elif value in {"--features", "-F"} and index + 1 < len(cargo_options):
|
|
raw = cargo_options[index + 1]
|
|
if raw:
|
|
features.update(part for part in re.split(r"[ ,]+", raw) if part)
|
|
|
|
target_triple: str | None = None
|
|
for index, value in enumerate(cargo_options):
|
|
if value.startswith("--target="):
|
|
target_triple = value.partition("=")[2]
|
|
elif value == "--target" and index + 1 < len(cargo_options):
|
|
target_triple = cargo_options[index + 1]
|
|
|
|
if all_features:
|
|
feature_signature = "all-features"
|
|
else:
|
|
parts = ["no-default" if no_default else "default"]
|
|
if features:
|
|
parts.append("features=" + ",".join(sorted(features)))
|
|
feature_signature = ";".join(parts)
|
|
|
|
# These disposable lanes intentionally contain a bounded family of
|
|
# package-specific feature sets. Hot reload needs the editor's
|
|
# `dev,hot-reload` build and `game_hot`'s `dylib` build in one isolated
|
|
# runtime lane; package aliases likewise expand to different internal
|
|
# features. They must never spill into the persistent default-feature lane.
|
|
if lane == "hot-reload":
|
|
if all_features:
|
|
raise LaneError("all-features belongs in the candidate lane")
|
|
feature_signature = "hot-reload-family"
|
|
elif lane == "package":
|
|
feature_signature = "package-family"
|
|
|
|
if lane == "dev" and (all_features or "hot-reload" in features):
|
|
raise LaneError(
|
|
"the persistent dev lane cannot be used for all-features or hot-reload"
|
|
)
|
|
if lane == "candidate" and subcommand is not None and not all_features:
|
|
raise LaneError("the candidate lane requires --all-features")
|
|
|
|
return CommandContext(
|
|
profile=profile,
|
|
profile_dir=profile_directory(profile),
|
|
feature_signature=feature_signature,
|
|
cargo_subcommand=subcommand,
|
|
target_triple=target_triple,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LaneLayout:
|
|
workspace: str
|
|
workspace_hash: str
|
|
lane: str
|
|
mode: str
|
|
cache_root: str
|
|
lane_root: str
|
|
target_dir: str
|
|
build_dir: str
|
|
profile: str
|
|
profile_dir: str
|
|
runtime_deps: str
|
|
feature_signature: str
|
|
toolchain: dict[str, Any]
|
|
linker: str | None
|
|
rustflags: list[str]
|
|
rustflags_source: str
|
|
rustc_wrapper: str | None
|
|
target_triple: str
|
|
|
|
|
|
def lane_layout(
|
|
root: Path,
|
|
lane: str,
|
|
workflow: Mapping[str, Any],
|
|
command: Sequence[str] = (),
|
|
*,
|
|
toolchain: Mapping[str, Any] | None = None,
|
|
) -> LaneLayout:
|
|
lane = validate_lane_name(lane)
|
|
root = canonical_path(root)
|
|
info = dict(toolchain or toolchain_info())
|
|
context = command_context(command, lane)
|
|
storage = workflow["build_storage"]
|
|
cache = canonical_path(cache_root(root, workflow))
|
|
partition = cache / workspace_hash(root)
|
|
lane_root = partition / lane
|
|
separate = bool(info.get("separate_build_dir"))
|
|
effective_target = context.target_triple or str(info.get("host", "unknown"))
|
|
if (
|
|
context.target_triple
|
|
and context.target_triple != info.get("host")
|
|
and not lane.startswith("cross-")
|
|
):
|
|
raise LaneError("cross-target Cargo commands require a cross-* disposable lane")
|
|
|
|
persistent = str(storage.get("persistent_lane", "dev"))
|
|
if separate:
|
|
if lane == persistent:
|
|
target = _configured_path(
|
|
root, str(storage.get("dev_target_dir", "target"))
|
|
)
|
|
else:
|
|
exceptional = _configured_path(
|
|
root, str(storage.get("exceptional_target_root", "target/lanes"))
|
|
)
|
|
target = exceptional / lane
|
|
build = lane_root
|
|
mode = "separate-build-dir"
|
|
else:
|
|
target = lane_root / "target"
|
|
build = target
|
|
mode = "external-target-dir-fallback"
|
|
|
|
target_config = _repo_target_config(root, effective_target)
|
|
profile_path = Path(context.profile_dir)
|
|
if context.target_triple:
|
|
profile_path = Path(context.target_triple) / profile_path
|
|
return LaneLayout(
|
|
workspace=str(root),
|
|
workspace_hash=workspace_hash(root),
|
|
lane=lane,
|
|
mode=mode,
|
|
cache_root=str(cache),
|
|
lane_root=str(lane_root),
|
|
target_dir=str(target),
|
|
build_dir=str(build),
|
|
profile=context.profile,
|
|
profile_dir=context.profile_dir,
|
|
runtime_deps=str(build / profile_path / "deps"),
|
|
feature_signature=context.feature_signature,
|
|
toolchain=info,
|
|
linker=target_config.get("linker"),
|
|
rustflags=list(target_config.get("rustflags", [])),
|
|
rustflags_source=str(target_config.get("rustflags_source")),
|
|
rustc_wrapper=target_config.get("rustc_wrapper"),
|
|
target_triple=effective_target,
|
|
)
|
|
|
|
|
|
def lane_environment(layout: LaneLayout) -> dict[str, str]:
|
|
environment = {
|
|
"CARGO_TARGET_DIR": layout.target_dir,
|
|
"BLACKSITE_CARGO_LANE": layout.lane,
|
|
"BLACKSITE_WORKSPACE_HASH": layout.workspace_hash,
|
|
}
|
|
if layout.mode == "separate-build-dir":
|
|
environment["CARGO_BUILD_BUILD_DIR"] = layout.build_dir
|
|
if layout.lane in {"candidate", "package", "full-debug"}:
|
|
environment["CARGO_INCREMENTAL"] = "0"
|
|
if layout.lane == "full-debug":
|
|
environment["CARGO_PROFILE_DEV_DEBUG"] = "full"
|
|
environment["CARGO_PROFILE_TEST_DEBUG"] = "full"
|
|
return environment
|
|
|
|
|
|
def sentinel_path(path: Path) -> Path:
|
|
return path / SENTINEL_NAME
|
|
|
|
|
|
def read_sentinel(path: Path) -> dict[str, Any] | None:
|
|
marker = sentinel_path(path)
|
|
if not marker.is_file() or marker.is_symlink():
|
|
return None
|
|
try:
|
|
with marker.open("r", encoding="utf-8") as handle:
|
|
value = json.load(handle)
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
def validate_sentinel(
|
|
path: Path,
|
|
document: Mapping[str, Any],
|
|
*,
|
|
root: Path,
|
|
lane: str,
|
|
role: str,
|
|
) -> list[str]:
|
|
errors: list[str] = []
|
|
expected = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"workspace": str(canonical_path(root)),
|
|
"workspace_hash": workspace_hash(root),
|
|
"lane": lane,
|
|
"role": role,
|
|
"path": str(canonical_path(path)),
|
|
}
|
|
for key, value in expected.items():
|
|
if document.get(key) != value:
|
|
errors.append(f"{key}: expected {value!r}, found {document.get(key)!r}")
|
|
return errors
|
|
|
|
|
|
def _signature(layout: LaneLayout) -> dict[str, Any]:
|
|
# The persistent development and candidate lanes intentionally serve both
|
|
# Cargo's `dev` and `test` profiles. Cargo fingerprints those profiles
|
|
# independently, while keeping them in one bounded ordinary/candidate lane
|
|
# avoids multiplying the Bevy dependency graph into permanent caches.
|
|
profile_signature = "dev-test" if layout.profile in {"dev", "test"} else layout.profile
|
|
return {
|
|
"toolchain": layout.toolchain,
|
|
"target_triple": layout.target_triple,
|
|
"linker": layout.linker,
|
|
"rustflags": layout.rustflags,
|
|
"rustc_wrapper": layout.rustc_wrapper,
|
|
"feature_signature": layout.feature_signature,
|
|
"profile_signature": profile_signature,
|
|
"mode": layout.mode,
|
|
"lane_environment": lane_environment(layout),
|
|
}
|
|
|
|
|
|
def _signature_compatible(existing: object, requested: Mapping[str, Any]) -> bool:
|
|
if existing == requested:
|
|
return True
|
|
if not isinstance(existing, Mapping):
|
|
return False
|
|
previous = dict(existing)
|
|
current = dict(requested)
|
|
previous_profile = previous.pop("profile_signature", None)
|
|
current_profile = current.pop("profile_signature", None)
|
|
return (
|
|
previous == current
|
|
and previous_profile in {"dev", "test"}
|
|
and current_profile == "dev-test"
|
|
)
|
|
|
|
|
|
def _atomic_json(path: Path, document: Mapping[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
descriptor, temporary = tempfile.mkstemp(
|
|
prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
|
|
)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
|
json.dump(document, handle, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _prepare_marker(
|
|
path: Path, layout: LaneLayout, role: str, *, allow_create: bool
|
|
) -> dict[str, Any]:
|
|
if path.is_symlink():
|
|
raise LaneError(f"refusing symlinked lane path: {path}")
|
|
if allow_create:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
if not path.is_dir():
|
|
raise LaneError(f"lane path does not exist: {path}")
|
|
|
|
existing = read_sentinel(path)
|
|
now = utc_now()
|
|
signature = _signature(layout)
|
|
if existing is not None:
|
|
errors = validate_sentinel(
|
|
path, existing, root=Path(layout.workspace), lane=layout.lane, role=role
|
|
)
|
|
if errors:
|
|
raise LaneError(
|
|
f"invalid lane sentinel at {path}: " + "; ".join(errors)
|
|
)
|
|
if not _signature_compatible(existing.get("signature"), signature):
|
|
raise LaneError(
|
|
"lane signature changed; select the correct exceptional lane or reset "
|
|
f"the whole lane after a dry run: {layout.lane}"
|
|
)
|
|
document = dict(existing)
|
|
document["signature"] = signature
|
|
document["last_use"] = now
|
|
document["last_profile"] = layout.profile
|
|
else:
|
|
document = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"workspace": layout.workspace,
|
|
"workspace_hash": layout.workspace_hash,
|
|
"lane": layout.lane,
|
|
"role": role,
|
|
"path": str(canonical_path(path)),
|
|
"target_dir": layout.target_dir,
|
|
"build_dir": layout.build_dir,
|
|
"signature": signature,
|
|
"created_at": now,
|
|
"last_use": now,
|
|
"last_profile": layout.profile,
|
|
}
|
|
_atomic_json(sentinel_path(path), document)
|
|
return document
|
|
|
|
|
|
def prepare_lane(layout: LaneLayout) -> None:
|
|
lane_root = Path(layout.lane_root)
|
|
_prepare_marker(lane_root, layout, "build", allow_create=True)
|
|
|
|
target = Path(layout.target_dir)
|
|
workspace_target = Path(layout.workspace) / "target"
|
|
if target != workspace_target and target != Path(layout.build_dir):
|
|
_prepare_marker(target, layout, "target", allow_create=True)
|
|
|
|
|
|
def require_prepared_lane(layout: LaneLayout) -> None:
|
|
lane_root = Path(layout.lane_root)
|
|
marker = read_sentinel(lane_root)
|
|
if marker is None:
|
|
raise LaneError(
|
|
f"lane {layout.lane!r} has not been built through cargo_lane.py"
|
|
)
|
|
errors = validate_sentinel(
|
|
lane_root,
|
|
marker,
|
|
root=Path(layout.workspace),
|
|
lane=layout.lane,
|
|
role="build",
|
|
)
|
|
if errors:
|
|
raise LaneError("invalid prepared lane: " + "; ".join(errors))
|
|
|
|
|
|
def layout_document(layout: LaneLayout) -> dict[str, Any]:
|
|
document = asdict(layout)
|
|
document["environment"] = lane_environment(layout)
|
|
document["sentinel"] = str(sentinel_path(Path(layout.lane_root)))
|
|
return document
|
|
|
|
|
|
def _shell_environment(values: Mapping[str, str]) -> str:
|
|
return "\n".join(
|
|
f"export {name}={shlex.quote(value)}" for name, value in sorted(values.items())
|
|
)
|
|
|
|
|
|
def _run_command(layout: LaneLayout, command: Sequence[str], *, dry_run: bool) -> int:
|
|
if not command:
|
|
raise LaneError("missing command after --")
|
|
if dry_run:
|
|
print(
|
|
json.dumps(
|
|
{"layout": layout_document(layout), "command": list(command)},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
prepare_lane(layout)
|
|
environment = os.environ.copy()
|
|
environment.update(lane_environment(layout))
|
|
return subprocess.run(command, cwd=layout.workspace, env=environment).returncode
|
|
|
|
|
|
def _resolve_runtime_command(layout: LaneLayout, command: Sequence[str]) -> list[str]:
|
|
if not command:
|
|
raise LaneError("missing runtime command after --")
|
|
resolved = list(command)
|
|
executable = Path(resolved[0])
|
|
if not executable.is_absolute():
|
|
parts = executable.parts
|
|
if parts and parts[0] == "target":
|
|
executable = Path(layout.target_dir).joinpath(*parts[1:])
|
|
elif "/" in resolved[0]:
|
|
executable = Path(layout.workspace) / executable
|
|
resolved[0] = str(canonical_path(executable)) if "/" in str(executable) else str(executable)
|
|
return resolved
|
|
|
|
|
|
def _run_runtime(layout: LaneLayout, command: Sequence[str], *, dry_run: bool) -> int:
|
|
require_prepared_lane(layout)
|
|
resolved = _resolve_runtime_command(layout, command)
|
|
environment = os.environ.copy()
|
|
dependencies = layout.runtime_deps
|
|
existing = environment.get("LD_LIBRARY_PATH")
|
|
environment["LD_LIBRARY_PATH"] = (
|
|
dependencies if not existing else dependencies + os.pathsep + existing
|
|
)
|
|
environment.update(lane_environment(layout))
|
|
if dry_run:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"layout": layout_document(layout),
|
|
"command": resolved,
|
|
"LD_LIBRARY_PATH": environment["LD_LIBRARY_PATH"],
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
os.chdir(layout.workspace)
|
|
os.execvpe(resolved[0], resolved, environment)
|
|
raise AssertionError("os.execvpe unexpectedly returned") # pragma: no cover
|
|
|
|
|
|
def _self_test() -> dict[str, Any]:
|
|
checks: list[str] = []
|
|
assert parse_version("cargo 1.97.0 (abc)") == (1, 97, 0)
|
|
checks.append("cargo-version-parsing")
|
|
assert profile_directory("test") == "debug"
|
|
assert profile_directory("package-qa") == "package-qa"
|
|
checks.append("profile-directory-mapping")
|
|
try:
|
|
validate_lane_name("../escape")
|
|
except LaneError:
|
|
checks.append("lane-name-traversal-rejected")
|
|
else: # pragma: no cover
|
|
raise AssertionError("unsafe lane name accepted")
|
|
|
|
with tempfile.TemporaryDirectory(prefix="blacksite-lane-self-test-") as temporary:
|
|
root = Path(temporary) / "workspace"
|
|
(root / ".git").mkdir(parents=True)
|
|
(root / "Cargo.toml").write_text("[workspace]\nmembers = []\n", encoding="utf-8")
|
|
workflow = {"build_storage": dict(DEFAULT_BUILD_STORAGE)}
|
|
workflow["build_storage"]["cache_root"] = str(Path(temporary) / "cache")
|
|
fake_toolchain = {
|
|
"cargo": "cargo 1.97.0",
|
|
"cargo_version": "1.97.0",
|
|
"rustc": "rustc 1.97.0",
|
|
"host": "x86_64-unknown-linux-gnu",
|
|
"separate_build_dir": True,
|
|
}
|
|
layout = lane_layout(root, "dev", workflow, ["cargo", "check"], toolchain=fake_toolchain)
|
|
prepare_lane(layout)
|
|
marker = read_sentinel(Path(layout.lane_root))
|
|
assert marker is not None
|
|
assert not validate_sentinel(
|
|
Path(layout.lane_root), marker, root=root, lane="dev", role="build"
|
|
)
|
|
checks.append("sentinel-round-trip")
|
|
assert Path(layout.target_dir) == root / "target"
|
|
assert Path(layout.build_dir) != Path(layout.target_dir)
|
|
checks.append("separate-build-dir-layout")
|
|
test_layout = lane_layout(
|
|
root, "dev", workflow, ["cargo", "test"], toolchain=fake_toolchain
|
|
)
|
|
prepare_lane(test_layout)
|
|
assert read_sentinel(Path(layout.lane_root))["signature"]["profile_signature"] == "dev-test"
|
|
checks.append("dev-and-test-share-persistent-lane")
|
|
hot_editor = lane_layout(
|
|
root,
|
|
"hot-reload",
|
|
workflow,
|
|
["cargo", "build", "-p", "editor", "--features", "dev,hot-reload"],
|
|
toolchain=fake_toolchain,
|
|
)
|
|
hot_game = lane_layout(
|
|
root,
|
|
"hot-reload",
|
|
workflow,
|
|
["cargo", "build", "-p", "game_hot", "--features", "dylib"],
|
|
toolchain=fake_toolchain,
|
|
)
|
|
assert hot_editor.feature_signature == hot_game.feature_signature == "hot-reload-family"
|
|
checks.append("hot-reload-feature-family-isolated")
|
|
package = lane_layout(
|
|
root,
|
|
"package",
|
|
workflow,
|
|
["cargo", "package-project", "--project", ".", "--profile", "package-qa"],
|
|
toolchain=fake_toolchain,
|
|
)
|
|
assert package.profile == "dev"
|
|
assert package.feature_signature == "package-family"
|
|
checks.append("project-alias-options-not-cargo-options")
|
|
|
|
fallback_toolchain = dict(fake_toolchain)
|
|
fallback_toolchain["cargo_version"] = "1.90.0"
|
|
fallback_toolchain["separate_build_dir"] = False
|
|
fallback = lane_layout(
|
|
root,
|
|
"candidate",
|
|
workflow,
|
|
["cargo", "test", "--all-features"],
|
|
toolchain=fallback_toolchain,
|
|
)
|
|
assert fallback.mode == "external-target-dir-fallback"
|
|
assert Path(fallback.target_dir) == Path(fallback.lane_root) / "target"
|
|
assert fallback.build_dir == fallback.target_dir
|
|
checks.append("external-target-dir-fallback")
|
|
|
|
cross = lane_layout(
|
|
root,
|
|
"cross-wasm",
|
|
workflow,
|
|
["cargo", "check", "--target", "wasm32-unknown-unknown"],
|
|
toolchain=fake_toolchain,
|
|
)
|
|
assert cross.target_triple == "wasm32-unknown-unknown"
|
|
assert "wasm32-unknown-unknown" in cross.runtime_deps
|
|
checks.append("cross-target-partition")
|
|
|
|
return {"ok": True, "checks": checks}
|
|
|
|
|
|
def _command_tail(values: Iterable[str]) -> list[str]:
|
|
result = list(values)
|
|
if result and result[0] == "--":
|
|
result.pop(0)
|
|
return result
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--workspace", type=Path, help="workspace root override")
|
|
subparsers = parser.add_subparsers(dest="action", required=True)
|
|
|
|
env_parser = subparsers.add_parser("env", help="describe a lane environment")
|
|
env_parser.add_argument("lane")
|
|
env_parser.add_argument("--json", action="store_true")
|
|
|
|
exec_parser = subparsers.add_parser("exec", help="execute a command in a lane")
|
|
exec_parser.add_argument("lane")
|
|
exec_parser.add_argument("--dry-run", action="store_true")
|
|
exec_parser.add_argument("command", nargs=argparse.REMAINDER)
|
|
|
|
run_parser = subparsers.add_parser("run", help="run an existing lane artifact")
|
|
run_parser.add_argument("lane")
|
|
run_parser.add_argument("--dry-run", action="store_true")
|
|
run_parser.add_argument("command", nargs=argparse.REMAINDER)
|
|
|
|
subparsers.add_parser("self-test", help="run temp-directory safety checks")
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
arguments = build_parser().parse_args(argv)
|
|
if arguments.action == "self-test":
|
|
print(json.dumps(_self_test(), indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
try:
|
|
root = workspace_root(arguments.workspace)
|
|
workflow = load_workflow(root)
|
|
raw_command = list(getattr(arguments, "command", []))
|
|
# argparse.REMAINDER intentionally preserves arbitrary child-command
|
|
# flags. Accept our one wrapper flag after the lane as well as before it.
|
|
if arguments.action in {"exec", "run"} and "--dry-run" in raw_command:
|
|
separator = raw_command.index("--") if "--" in raw_command else len(raw_command)
|
|
if raw_command.index("--dry-run") < separator:
|
|
arguments.dry_run = True
|
|
raw_command.remove("--dry-run")
|
|
command = _command_tail(raw_command)
|
|
layout = lane_layout(root, arguments.lane, workflow, command)
|
|
|
|
if arguments.action == "env":
|
|
document = layout_document(layout)
|
|
if arguments.json:
|
|
print(json.dumps(document, indent=2, sort_keys=True))
|
|
else:
|
|
print(_shell_environment(document["environment"]))
|
|
return 0
|
|
if arguments.action == "exec":
|
|
return _run_command(layout, command, dry_run=arguments.dry_run)
|
|
if arguments.action == "run":
|
|
return _run_runtime(layout, command, dry_run=arguments.dry_run)
|
|
except LaneError as error:
|
|
print(f"cargo-lane: {error}", file=sys.stderr)
|
|
return 2
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|