diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 502d503..01cb0df 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.7.0" +VERSION = "1.8.0" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -42,6 +42,10 @@ MANIFEST_MAX_SECONDS = 15 MANIFEST_MAX_DEPTH = 64 LIBRARY_DOWNLOAD_WORKERS = 4 HEALTH_LOCK_TIMEOUT_SECONDS = 5 +PLUTONIUM_MANIFEST_URL = "https://cdn.plutoniummod.com/updater/prod/info.json" +PLUTONIUM_MANIFEST_LIMIT = 2 * 1024 * 1024 +PLUTONIUM_FILE_LIMIT = 512 * 1024 * 1024 +PLUTONIUM_CACHE_LIMIT = 4 * 1024 * 1024 * 1024 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") AGENT_CONFIG = Path("/etc/electric-cherry/agent.json") AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py") @@ -487,6 +491,34 @@ def managed_write(root, value, content, uid=1000, gid=1000): os.unlink(temp_name, dir_fd=parent) +def managed_copy(root, value, source, uid=1000, gid=1000): + """Copy a private file, using Linux reflinks when supported, never hardlinks.""" + with managed_open(Path(source).parent, Path(source).name) as incoming: + with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name): + temp_name = ".ec-" + uuid.uuid4().hex + fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent) + try: + os.fchmod(fd, 0o644) + if os.geteuid() == 0: + os.fchown(fd, uid, gid) + try: + if not sys.platform.startswith("linux"): + raise OSError("Reflinks unavailable") + fcntl.ioctl(fd, 0x40049409, incoming.fileno()) # FICLONE + except OSError: + os.ftruncate(fd, 0) + with os.fdopen(fd, "wb", closefd=False) as outgoing: + shutil.copyfileobj(incoming, outgoing, 1024 * 1024) + outgoing.flush() + os.fsync(fd) + os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent) + os.fsync(parent) + finally: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + os.unlink(temp_name, dir_fd=parent) + + class ReadWriteGate: """Let observations run together; global mutations wait for existing readers.""" def __init__(self): @@ -921,11 +953,13 @@ class Agent: self.read_claims = set() self.health_probe_gate = threading.BoundedSemaphore(2) self.cancelled_creations = frozenset() + self.defer_plutonium_preparation = False self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.library_progress = {} self.library_transfers = {} self.inventory_manifest_cache = {} self.inventory_manifest_lock = threading.Lock() + self.plutonium_lock = threading.RLock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None self.lane_lock = threading.Lock() @@ -1360,8 +1394,9 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) - return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} + capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) + plutonium_cache = self.plutonium_cache_inventory() + return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} def safe_to_disrupt(self, instance_id, generation): health = self.inspect(instance_id, generation) @@ -1510,6 +1545,223 @@ class Agent: raise Rejected("Initial files exceed the 64 MiB operation limit") return game, library, image, port, bind, env + @staticmethod + def plutonium_manifest(data): + """Validate the official updater format before using any paths or URLs.""" + if len(data) > PLUTONIUM_MANIFEST_LIMIT: + raise AgentError("Plutonium manifest is too large") + try: + manifest = json.loads(data) + except (ValueError, UnicodeError): + raise AgentError("Invalid Plutonium manifest") from None + if not isinstance(manifest, dict) or manifest.get("product") != "plutonium-core-prod": + raise AgentError("Invalid Plutonium production manifest") + revision = manifest.get("revision") + if type(revision) is not int or not 1 <= revision <= 2147483647: + raise AgentError("Invalid Plutonium revision") + base = manifest.get("baseUrl") + if not isinstance(base, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/(?:[A-Za-z0-9_-]+/)*", base): + raise AgentError("Plutonium downloads must use the official production CDN") + entries = manifest.get("files") + if not isinstance(entries, list) or not 1 <= len(entries) <= 4096: + raise AgentError("Invalid Plutonium file list") + names, total = set(), 0 + for entry in entries: + if not isinstance(entry, dict): + raise AgentError("Invalid Plutonium file entry") + name, size, checksum = entry.get("name"), entry.get("size"), entry.get("hash") + if not isinstance(name, str) or len(name) > 512 or re.search(r'[\x00-\x1f\x7f\\:*?"<>|]', name): + raise AgentError("Invalid Plutonium file path") + path = PurePosixPath(name) + if not path.parts or path.is_absolute() or str(path) != name or any(part in (".", "..") or part.endswith((".", " ")) for part in path.parts) or name.lower() == "cdn_info.json": + raise AgentError("Invalid Plutonium file path") + if name.lower() in names: + raise AgentError("Duplicate Plutonium file path") + names.add(name.lower()) + if type(size) is not int or not 0 <= size <= PLUTONIUM_FILE_LIMIT: + raise AgentError("Invalid Plutonium file size") + if not isinstance(checksum, str) or not re.fullmatch(r"[a-f0-9]{40}", checksum): + raise AgentError("Invalid Plutonium file hash") + total += size + if total > PLUTONIUM_CACHE_LIMIT: + raise AgentError("Plutonium cache exceeds the size limit") + for name in names: + if any(str(parent) in names for parent in PurePosixPath(name).parents if str(parent) != "."): + raise AgentError("Conflicting Plutonium file paths") + if "bin/plutonium-bootstrapper-win32.exe" not in {entry["name"] for entry in entries}: + raise AgentError("Plutonium manifest is missing the server bootstrapper") + # Match plutonium-updater's default: launcher assets are optional. + return manifest, [entry for entry in entries if not entry["name"].startswith("launcher")] + + @staticmethod + def plutonium_file_matches(root, entry): + try: + with managed_open(root, entry["name"]) as stream: + if os.fstat(stream.fileno()).st_size != entry["size"]: + return False + checksum = hashlib.sha1() + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + checksum.update(chunk) + return checksum.hexdigest() == entry["hash"] + except (OSError, AgentError): + return False + + def plutonium_open(self, url): + if not isinstance(url, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/[A-Za-z0-9_./-]+", url): + raise AgentError("Plutonium downloads must use the official production CDN") + class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, hdrs, newurl): + raise AgentError("Plutonium download redirects are not allowed") + # Never send the controller's host credentials to the public CDN. + return urllib.request.build_opener(NoRedirect()).open(url, timeout=30) + + def plutonium_snapshot(self): + root = scoped(self.root / "agent", "plutonium-cache") + pointer = json.loads(managed_read(root, "current.json") or b"null") + if not isinstance(pointer, dict) or not isinstance(pointer.get("key"), str) or not re.fullmatch(r"[a-f0-9]{64}", pointer["key"]): + raise AgentError("Plutonium cache is not prepared") + content = scoped(root, pointer["key"] + "/content") + raw = managed_read(content, "cdn_info.json") + if raw is None or digest(raw) != pointer["key"]: + raise AgentError("Plutonium cache manifest is invalid") + manifest, files = self.plutonium_manifest(raw) + if pointer.get("revision") != manifest["revision"] or pointer.get("bytes") != sum(entry["size"] for entry in files): + raise AgentError("Plutonium cache metadata is invalid") + # Heartbeat checks are cheap. Every file is hashed before it is seeded. + for entry in files: + info = scoped(content, entry["name"]).lstat() + if not stat_module.S_ISREG(info.st_mode) or info.st_size != entry["size"]: + raise AgentError("Plutonium cache files are incomplete") + return pointer, content, raw, files + + def plutonium_cache_inventory(self): + with self.plutonium_lock: + try: + pointer, _, _, _ = self.plutonium_snapshot() + return {"revision": pointer["revision"], "bytes": pointer["bytes"]} + except (OSError, ValueError, AgentError): + return None + + def prepare_plutonium(self, payload): + target, reserve = payload.get("revision", 0), payload.get("minimumFreeBytes", 0) + if type(target) is not int or not 0 <= target <= 2147483647 or type(reserve) is not int or reserve < 0: + raise Rejected("Invalid Plutonium preparation request") + root = scoped(self.root / "agent", "plutonium-cache") + root.mkdir(mode=0o700, exist_ok=True) + os.chmod(root, 0o700) + deadline = time.monotonic() + 1800 + def active(): + if self.stopping.is_set() or time.monotonic() >= deadline: + raise AgentError("Plutonium cache preparation was interrupted or timed out") + if self.defer_plutonium_preparation: + raise Held("Plutonium cache preparation is waiting for other host operations") + active() + try: + with self.plutonium_open(PLUTONIUM_MANIFEST_URL) as response: + raw = response.read(PLUTONIUM_MANIFEST_LIMIT + 1) + manifest, files = self.plutonium_manifest(raw) + if manifest["revision"] < target: + raise AgentError("Plutonium CDN has not reached the requested revision") + key = digest(raw) + previous = None + with self.plutonium_lock: + with contextlib.suppress(OSError, ValueError, AgentError): + _, previous, _, _ = self.plutonium_snapshot() + if previous is not None and previous.parent.name == key and all(self.plutonium_file_matches(previous, entry) for entry in files): + return {"revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)} + staging = scoped(root, key + ".partial/content") + staging.mkdir(parents=True, exist_ok=True) + expected = {entry["name"] for entry in files} + # SIGKILL/power loss bypasses download/copy finally blocks. Remove + # abandoned temporary files before counting space or publishing. + for directory, subdirs, filenames in os.walk(staging, followlinks=False): + for name in subdirs + filenames: + path = Path(directory) / name + if path.is_symlink() or (name in filenames and path.relative_to(staging).as_posix() not in expected): + path.unlink() + # Keep at most the current snapshot and this resumable preparation. + current_key = previous.parent.name if previous is not None else None + for directory in root.iterdir(): + if re.fullmatch(r"[a-f0-9]{64}(?:\.partial)?", directory.name) and directory.name not in (current_key, key + ".partial"): + if directory.is_symlink(): + directory.unlink() + else: + shutil.rmtree(directory) + missing = sum(entry["size"] for entry in files if not self.plutonium_file_matches(staging, entry)) + if shutil.disk_usage(root).free < reserve + missing + len(raw): + raise AgentError("Insufficient free storage for the Plutonium cache and host reserve") + for entry in files: + active() + if self.plutonium_file_matches(staging, entry): + continue + if previous is not None and self.plutonium_file_matches(previous, entry): + managed_copy(staging, entry["name"], scoped(previous, entry["name"]), os.geteuid(), os.getegid()) + continue + destination = scoped(staging, entry["name"]) + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_name(".download-" + uuid.uuid4().hex) + try: + with self.plutonium_open(manifest["baseUrl"] + entry["hash"]) as incoming, temporary.open("xb") as outgoing: + size, checksum = 0, hashlib.sha1() + while True: + active() + chunk = incoming.read(min(1024 * 1024, entry["size"] - size + 1)) + if not chunk: + break + size += len(chunk) + if size > entry["size"]: + raise AgentError("Plutonium download exceeds its expected size") + checksum.update(chunk) + outgoing.write(chunk) + if size != entry["size"] or checksum.hexdigest() != entry["hash"]: + raise AgentError("Plutonium download failed size or checksum verification") + outgoing.flush() + os.fsync(outgoing.fileno()) + os.replace(temporary, destination) + sync_parent(destination) + finally: + with contextlib.suppress(FileNotFoundError): + temporary.unlink() + # Publish metadata last. Interrupted transfers can never claim ready. + active() + atomic_bytes(staging / "cdn_info.json", raw) + pointer = {"key": key, "revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)} + with self.plutonium_lock: + final = scoped(root, key) + if final.exists(): + # Invalidate before replacing a damaged/same-revision cache. + if current_key == key: + (root / "current.json").unlink(missing_ok=True) + shutil.rmtree(final) + os.replace(staging.parent, final) + sync_parent(final) + atomic_json(root / "current.json", pointer) + if previous is not None and current_key != key: + shutil.rmtree(previous.parent) + return {"revision": pointer["revision"], "bytes": pointer["bytes"]} + except (urllib.error.URLError, TimeoutError, http.client.HTTPException): + raise AgentError("Plutonium CDN download is unavailable; the previous cache is retained") from None + + def seed_plutonium(self, app, image): + if not re.fullmatch(r"ghcr\.io/ayymoss/plutainer(?::[A-Za-z0-9_.-]+)?(?:@sha256:[a-f0-9]{64})?", image): + return False + with self.plutonium_lock: + try: + _, content, raw, files = self.plutonium_snapshot() + if not all(self.plutonium_file_matches(content, entry) for entry in files): + scoped(self.root / "agent", "plutonium-cache/current.json").unlink(missing_ok=True) + raise AgentError("Plutonium cache checksum verification failed") + except (OSError, ValueError, AgentError): + return False + uid, gid = int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)) + marker = "runtime/plutonium/cdn_info.json" + managed_write(app, marker, None, uid, gid) + for entry in files: + self.check_creation_cancellation() + managed_copy(app, "runtime/plutonium/" + entry["name"], scoped(content, entry["name"]), uid, gid) + managed_write(app, marker, raw, uid, gid) + return True + def create(self, instance_id, generation, payload): self.check_creation_cancellation() directory = self.instance_dir(instance_id, generation) @@ -1527,6 +1779,7 @@ class Agent: app = directory / "app" app.mkdir(parents=True, exist_ok=True) self.make_writable(app, directory=True) + self.seed_plutonium(app, image) self.write_initial_files(app, payload.get("files", [])) atomic_json(directory / "spec.json", payload) env_path = directory / "environment" @@ -2184,6 +2437,8 @@ class Agent: return {"image": image, "ready": True} if kind == "host.inspect": return self.inventory() + if kind == "plutonium.prepare": + return self.prepare_plutonium(payload) if kind == "rcon.status-batch": return self.rcon_status_batch(operation) if kind == "library.download": @@ -2290,7 +2545,7 @@ class Agent: if not self.config.get("pendingCredential") or self.recover_credential(): return self.journal.update(operation["id"], "completed", {"rotated": True, "reconciled": True}) return row - if kind == "runtime.pull": + if kind in ("runtime.pull", "plutonium.prepare"): # Docker resumes already downloaded layers. Pull again under # the same operation ID: an existing tag can still refer to an # older image and cannot prove an interrupted pull completed. @@ -2298,6 +2553,8 @@ class Agent: try: result = self.dispatch(operation) return self.journal.update(operation["id"], "completed", {**result, "reconciled": True}) + except Held as err: + return self.journal.update(operation["id"], "accepted", {"blocked": "host-busy", "message": str(err)}) except Exception: return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed") if kind == "library.import": @@ -2407,6 +2664,7 @@ class Agent: # Only the heartbeat thread replaces this immutable snapshot; it never # mutates the operation journal or races Docker cleanup against create. self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str)) + self.defer_plutonium_preparation = response.get("deferPlutoniumPreparation") is True keep_days = response.get("observationKeepDays") self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0 host = response.get("host", {}) diff --git a/agent/tests/test_plutonium_cache.py b/agent/tests/test_plutonium_cache.py new file mode 100644 index 0000000..99f77d3 --- /dev/null +++ b/agent/tests/test_plutonium_cache.py @@ -0,0 +1,506 @@ +import base64 +import copy +import hashlib +import json +from pathlib import Path +import unittest +from unittest.mock import patch +import urllib.error + +import test_agent as fixtures + + +module = fixtures.module +AgentError = fixtures.AgentError +IMAGE = "ghcr.io/ayymoss/plutainer:latest" +BOOTSTRAPPER = "bin/plutonium-bootstrapper-win32.exe" +BASE_URL = "https://cdn.plutoniummod.com/updater/prod/files/" + + +class PlutoniumCacheTests(unittest.TestCase): + tearDown = fixtures.AgentTests.tearDown + operation = fixtures.AgentTests.operation + create = fixtures.AgentTests.create + + def setUp(self): + fixtures.AgentTests.setUp(self) + network = patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Unexpected vendor request"), create=True) + network.start() + self.addCleanup(network.stop) + + def release(self, revision=5354, contents=None): + contents = contents or { + BOOTSTRAPPER: b"bootstrapper", + "bin/plutonium.dll": b"runtime", + "storage/t5/zone/plutonium.ff": b"zone-data", + "storage/iw5/images/~weapon_ak47_iw5_spc-rgb&weap~fa48e615.iwi": b"texture-data", + "launcher/plutonium.exe": b"launcher-not-needed", + } + manifest = { + "product": "plutonium-core-prod", + "revision": revision, + "baseUrl": BASE_URL, + "files": [ + {"name": name, "size": len(content), "hash": hashlib.sha1(content).hexdigest()} + for name, content in contents.items() + ], + } + return manifest, contents + + def opener(self, manifest, contents, calls=None): + calls = calls if calls is not None else [] + + def open_url(url): + calls.append(url) + if url == module.PLUTONIUM_MANIFEST_URL: + return fixtures.Response(json.dumps(manifest).encode()) + self.assertTrue(url.startswith(BASE_URL), url) + name = next(entry["name"] for entry in manifest["files"] if url == BASE_URL + entry["hash"]) + return fixtures.Response(contents[name]) + + return open_url + + def file_url(self, content): + return BASE_URL + hashlib.sha1(content).hexdigest() + + def prepare(self, revision=5354, contents=None): + manifest, contents = self.release(revision, contents) + calls = [] + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + result = self.agent.prepare_plutonium({"revision": revision}) + return manifest, contents, calls, result + + def cache_content(self): + root = self.agent.root / "agent" / "plutonium-cache" + pointer = json.loads((root / "current.json").read_text()) + return root / pointer["key"] / "content" + + def app(self, name="app"): + app = Path(self.temp.name) / name + app.mkdir(exist_ok=True) + return app + + def assert_runtime(self, app, manifest, contents): + runtime = app / "runtime" / "plutonium" + self.assertEqual(json.loads((runtime / "cdn_info.json").read_text()), manifest) + for name, content in contents.items(): + if name.startswith("launcher"): + self.assertFalse((runtime / name).exists()) + else: + self.assertEqual((runtime / name).read_bytes(), content) + + def test_cold_preparation_verifies_downloads_and_reports_inventory(self): + self.assertIsNone(self.agent.plutonium_cache_inventory()) + manifest, contents, calls, result = self.prepare() + expected_bytes = sum(len(content) for name, content in contents.items() if not name.startswith("launcher")) + self.assertEqual(result, {"revision": 5354, "bytes": expected_bytes}) + self.assertEqual(self.agent.plutonium_cache_inventory(), result) + self.assertEqual(calls[0], module.PLUTONIUM_MANIFEST_URL) + self.assertEqual(set(calls[1:]), {self.file_url(content) for name, content in contents.items() if not name.startswith("launcher")}) + self.assertEqual(json.loads((self.cache_content() / "cdn_info.json").read_text()), manifest) + self.assertEqual((self.agent.root / "agent" / "plutonium-cache").stat().st_mode & 0o077, 0) + + def test_warm_preparation_only_fetches_manifest_and_dispatches(self): + manifest, contents, _, first = self.prepare() + calls = [] + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + result = self.agent.dispatch({"kind": "plutonium.prepare", "payload": {"revision": 5354}}) + self.assertEqual(result, first) + self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) + + def test_refresh_downloads_changed_files_and_reuses_verified_unchanged_files(self): + _, contents, _, _ = self.prepare() + changed = {**contents, "bin/plutonium.dll": b"new-runtime", "storage/t6/new.ff": b"new-zone"} + manifest, changed, calls, result = self.prepare(5355, changed) + self.assertEqual(result["revision"], 5355) + self.assertEqual(set(calls[1:]), {self.file_url(changed["bin/plutonium.dll"]), self.file_url(changed["storage/t6/new.ff"])}) + app = self.app() + self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) + self.assert_runtime(app, manifest, changed) + + def test_changed_manifest_is_refreshed_even_when_revision_number_is_unchanged(self): + _, contents, _, _ = self.prepare() + previous = self.cache_content() + manifest, changed, calls, result = self.prepare(5354, {**contents, "bin/plutonium.dll": b"corrected-runtime"}) + self.assertEqual(result["revision"], 5354) + self.assertNotEqual(self.cache_content(), previous) + self.assertEqual(calls[1:], [self.file_url(changed["bin/plutonium.dll"])]) + app = self.app() + self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) + self.assert_runtime(app, manifest, changed) + + def test_failed_refresh_preserves_the_older_complete_cache(self): + manifest, contents, _, result = self.prepare() + previous = self.cache_content() + next_manifest, next_contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) + next_contents["bin/plutonium.dll"] = b"bad-runtime" + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(next_manifest, next_contents)): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({"revision": 5355}) + self.assertEqual(self.cache_content(), previous) + self.assertEqual(self.agent.plutonium_cache_inventory(), result) + app = self.app() + self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) + self.assert_runtime(app, manifest, contents) + + def test_interrupted_preparation_resumes_without_advertising_a_partial_cache(self): + manifest, contents = self.release() + open_url = self.opener(manifest, contents) + + def interrupted_open(url): + if url == self.file_url(contents["bin/plutonium.dll"]): + raise urllib.error.URLError("simulated network interruption") + return open_url(url) + + with patch.object(self.agent, "plutonium_open", side_effect=interrupted_open): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({}) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + self.assertFalse((self.agent.root / "agent/plutonium-cache/current.json").exists()) + _, _, calls, result = self.prepare() + self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) + self.assertEqual(result["revision"], manifest["revision"]) + + def test_agent_restart_reconciles_preparation_instead_of_accepting_an_older_cache(self): + _, contents, _, _ = self.prepare() + operation = {"id": "prepare-new-release", "kind": "plutonium.prepare", "payload": {"revision": 5355}} + self.agent.journal.accept(operation) + self.agent.journal.update(operation["id"], "running") + self.agent.journal.close() + self.agent = fixtures.Agent(self.config, self.docker) + self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown") + manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) + calls = [] + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "completed", result) + self.assertEqual(json.loads(result["result"])["revision"], 5355) + self.assertTrue(json.loads(result["result"])["reconciled"]) + self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5355) + self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL, self.file_url(contents["bin/plutonium.dll"])]) + with patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Completed operation repeated")): + self.assertEqual(self.agent.execute(operation)["result"], result["result"]) + + def test_heartbeat_defers_preparation_only_for_explicit_true(self): + self.assertFalse(self.agent.defer_plutonium_preparation) + for response, expected in ( + ({"deferPlutoniumPreparation": True}, True), + ({}, False), + ({"deferPlutoniumPreparation": False}, False), + ({"deferPlutoniumPreparation": 1}, False), + ({"deferPlutoniumPreparation": "true"}, False), + ({"deferPlutoniumPreparation": None}, False), + ): + self.agent.defer_plutonium_preparation = not expected + with self.subTest(response=response), patch.object(self.agent, "request", return_value=response), patch.object(self.agent, "inventory", return_value={}), patch.object(self.agent, "record_health"): + self.agent.heartbeat() + self.assertIs(self.agent.defer_plutonium_preparation, expected) + + def assert_preparation_yields_and_resumes(self, *, restarted): + manifest, contents = self.release() + operation = {"id": "cooperative-prepare", "kind": "plutonium.prepare", "payload": {"revision": 5354}} + if restarted: + self.agent.journal.accept(operation) + self.agent.journal.update(operation["id"], "running") + self.agent.journal.close() + self.agent = fixtures.Agent(self.config, self.docker) + self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown") + agent = self.agent + open_url = self.opener(manifest, contents) + + class BusyResponse(fixtures.Response): + def read(self, size=-1): + chunk = super().read(size) + if chunk: + agent.defer_plutonium_preparation = True + return chunk + + def become_busy_during_download(url): + if url == self.file_url(contents["bin/plutonium.dll"]): + return BusyResponse(contents["bin/plutonium.dll"]) + return open_url(url) + + with patch.object(self.agent, "plutonium_open", side_effect=become_busy_during_download): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "accepted", result) + self.assertIn("waiting for other host operations", json.loads(result["result"])["message"]) + self.assertIsNone(self.agent.executing_id) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + self.assertTrue(self.agent.defer_plutonium_preparation) + # A yielded preparation must let other mutations finish before retrying. + self.create() + self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"]) + self.agent.defer_plutonium_preparation = False + calls = [] + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "completed", result) + self.assertEqual(result["id"], operation["id"]) + self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) + self.assertIn(self.file_url(contents["bin/plutonium.dll"]), calls) + self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5354) + + def test_busy_host_yields_mid_download_and_resumes_completed_files(self): + self.assert_preparation_yields_and_resumes(restarted=False) + + def test_reconciled_preparation_yields_mid_download_and_resumes_same_operation(self): + self.assert_preparation_yields_and_resumes(restarted=True) + + def test_resumption_removes_orphans_before_storage_check_and_publication(self): + manifest, contents = self.release() + raw = json.dumps(manifest).encode() + staging = self.agent.root / "agent/plutonium-cache" / (module.digest(raw) + ".partial/content") + bootstrapper = staging / BOOTSTRAPPER + bootstrapper.parent.mkdir(parents=True) + bootstrapper.write_bytes(contents[BOOTSTRAPPER]) + orphans = [staging / ".download-abandoned", staging / "bin/.ec-abandoned", staging / "unlisted/extra.dat", staging / "cdn_info.json"] + for orphan in orphans: + orphan.parent.mkdir(parents=True, exist_ok=True) + orphan.write_bytes(b"orphaned bytes") + outside = Path(self.temp.name) / "outside-cache" + outside.mkdir() + (outside / "untouched").write_bytes(b"private outside bytes") + link = staging / "outside-link" + link.symlink_to(outside, target_is_directory=True) + original_disk_usage = module.shutil.disk_usage + + def checked_storage(path): + for orphan in orphans: + self.assertFalse(orphan.exists(), "orphan must be removed before storage accounting") + self.assertFalse(link.is_symlink()) + return original_disk_usage(path) + + calls = [] + with patch.object(module.shutil, "disk_usage", side_effect=checked_storage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + result = self.agent.prepare_plutonium({}) + content = self.cache_content() + expected = {name for name in contents if not name.startswith("launcher")} + actual = {path.relative_to(content).as_posix() for path in content.rglob("*") if path.is_file()} + self.assertEqual(actual, expected | {"cdn_info.json"}) + self.assertEqual(result["bytes"], sum(len(contents[name]) for name in expected)) + self.assertEqual(sum(path.stat().st_size for path in content.rglob("*") if path.is_file()), result["bytes"] + len(raw)) + self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) + self.assertEqual((outside / "untouched").read_bytes(), b"private outside bytes") + app = self.app() + self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) + self.assert_runtime(app, manifest, contents) + self.assertFalse((app / "runtime/plutonium/unlisted/extra.dat").exists()) + + def test_public_downloads_reject_other_origins_and_never_send_host_credentials(self): + # Call the class method to bypass the fixture's default network blocker; + # the underlying opener is mocked, so no request reaches the network. + with patch.object(module.urllib.request, "build_opener") as build: + for url in ( + "http://cdn.plutoniummod.com/updater/prod/info.json", + "https://attacker.example/updater/prod/info.json", + "https://cdn.plutoniummod.com.attacker.example/updater/prod/info.json", + "https://secret@cdn.plutoniummod.com/updater/prod/info.json", + "https://cdn.plutoniummod.com/updater/dev/info.json", + ): + with self.subTest(url=url), self.assertRaises(AgentError): + fixtures.Agent.plutonium_open(self.agent, url) + build.assert_not_called() + fixtures.Agent.plutonium_open(self.agent, module.PLUTONIUM_MANIFEST_URL) + build.return_value.open.assert_called_once_with(module.PLUTONIUM_MANIFEST_URL, timeout=30) + handler = build.call_args.args[0] + for redirect in (module.PLUTONIUM_MANIFEST_URL, "https://attacker.example/collect"): + with self.subTest(redirect=redirect), self.assertRaises(AgentError): + handler.redirect_request(None, None, 302, "Found", {}, redirect) + + def test_rejects_revision_downgrade_and_invalid_preparation_arguments(self): + self.prepare() + manifest, contents = self.release(5353) + with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({"revision": 5354}) + for payload in ({"revision": -1}, {"revision": True}, {"revision": "5354"}, + {"minimumFreeBytes": -1}, {"minimumFreeBytes": True}, {"minimumFreeBytes": "10"}): + with self.subTest(payload=payload), self.assertRaises(AgentError): + self.agent.prepare_plutonium(payload) + + def test_disk_reserve_failure_leaves_existing_cache_usable(self): + _, contents, _, result = self.prepare() + manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) + calls = [] + usage = type(module.shutil.disk_usage(self.agent.root))(10000, 9900, 100) + with patch.object(module.shutil, "disk_usage", return_value=usage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({"minimumFreeBytes": 1000}) + self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) + self.assertEqual(self.agent.plutonium_cache_inventory(), result) + + def test_malformed_manifest_never_publishes_a_cache(self): + good, contents = self.release() + malformed = [ + None, [], {}, + {**good, "product": "other-product"}, + {**good, "revision": 0}, + {**good, "revision": True}, + {**good, "revision": "5354"}, + {**good, "baseUrl": "http://cdn.plutoniummod.com/updater/prod/"}, + {**good, "baseUrl": "https://attacker.example/"}, + {**good, "files": []}, + {**good, "files": {}}, + {**good, "files": good["files"][1:]}, + {**good, "files": good["files"] + [good["files"][0]]}, + ] + for manifest in malformed: + with self.subTest(manifest=manifest), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({}) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + + def test_manifest_rejects_unsafe_paths_sizes_and_hashes_before_downloading(self): + good, contents = self.release() + bad_entries = [ + {"name": name} for name in ( + ".", "../escape", "/absolute", "bin/../../escape", "bin//file", "bin/./file", + "bin\\file", "C:/escape", "bin/file\x00", "cdn_info.json", + ) + ] + [ + {"size": size} for size in (-1, True, "5", None) + ] + [ + {"hash": value} for value in ("", "0" * 39, "z" * 40, None) + ] + for override in bad_entries: + manifest = copy.deepcopy(good) + manifest["files"][1].update(override) + calls = [] + with self.subTest(override=override), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({}) + self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + + def test_short_oversize_and_wrong_hash_downloads_are_not_published(self): + manifest, contents = self.release() + for body in (b"short", contents[BOOTSTRAPPER] + b"extra", b"X" * len(contents[BOOTSTRAPPER])): + with self.subTest(body=body), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, {**contents, BOOTSTRAPPER: body})): + with self.assertRaises(AgentError): + self.agent.prepare_plutonium({}) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + + def test_seeds_are_private_and_preserve_existing_lobby_state(self): + manifest, contents, _, _ = self.prepare() + first, second = self.app("first"), self.app("second") + private = { + "configs/server.cfg": b"secret config", + "runtime/plutonium/plugins/private.dll": b"private plugin", + "runtime/plutonium/storage/t5/logs/games_mp.log": b"lobby log", + } + for name, body in private.items(): + target = first / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(body) + self.assertTrue(self.agent.seed_plutonium(first, IMAGE)) + self.assertTrue(self.agent.seed_plutonium(second, IMAGE)) + self.assert_runtime(first, manifest, contents) + self.assert_runtime(second, manifest, contents) + for name, body in private.items(): + self.assertEqual((first / name).read_bytes(), body) + self.assertFalse((second / name).exists()) + first_file = first / "runtime/plutonium" / BOOTSTRAPPER + second_file = second / "runtime/plutonium" / BOOTSTRAPPER + cached_file = self.cache_content() / BOOTSTRAPPER + self.assertNotEqual(first_file.stat().st_ino, second_file.stat().st_ino) + self.assertNotEqual(first_file.stat().st_ino, cached_file.stat().st_ino) + first_file.write_bytes(b"container-updated") + self.assertEqual(second_file.read_bytes(), contents[BOOTSTRAPPER]) + self.assertEqual(cached_file.read_bytes(), contents[BOOTSTRAPPER]) + + def test_absent_or_corrupt_cache_and_unsupported_image_fall_back(self): + app = self.app() + self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) + self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) + self.prepare() + for image in ("example.invalid/plutainer:latest", "ghcr.io/ayymoss/plutainer-other:latest", "ghcr.io/attacker/plutainer:latest"): + with self.subTest(image=image): + self.assertFalse(self.agent.seed_plutonium(app, image)) + (self.cache_content() / BOOTSTRAPPER).write_bytes(b"broken-cache") + self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) + self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) + self.assertIsNone(self.agent.plutonium_cache_inventory()) + + def test_interrupted_seed_does_not_install_revision_marker_and_can_retry(self): + manifest, contents, _, _ = self.prepare() + app = self.app() + marker = app / "runtime/plutonium/cdn_info.json" + marker.parent.mkdir(parents=True) + marker.write_text(json.dumps({"revision": 5353})) + copy_file = module.managed_copy + copied = [] + + def interrupted_copy(root, name, source, *args, **kwargs): + self.assertFalse(marker.exists(), "revision marker must be installed only after all files") + if copied: + raise OSError("simulated full disk") + copied.append(name) + return copy_file(root, name, source, *args, **kwargs) + + with patch.object(module, "managed_copy", side_effect=interrupted_copy): + with self.assertRaises((AgentError, OSError)): + self.agent.seed_plutonium(app, IMAGE) + self.assertFalse(marker.exists()) + self.assertTrue(copied) + self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) + self.assert_runtime(app, manifest, contents) + + def test_failed_seed_prevents_container_start(self): + self.prepare() + with patch.object(module, "managed_copy", side_effect=OSError("simulated full disk")): + payload = {"game": "t5", "libraryId": self.library_id, "image": IMAGE, "port": 28960, "env": {}} + result = self.agent.execute(self.operation("instance.create", payload, "failed-seed")) + self.assertEqual(result["state"], "failed", result) + self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls)) + self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists()) + + def test_symlink_in_cache_cannot_seed_outside_data(self): + _, contents, _, _ = self.prepare() + cached = self.cache_content() / BOOTSTRAPPER + outside = Path(self.temp.name) / "outside" + outside.write_bytes(contents[BOOTSTRAPPER]) + cached.unlink() + cached.symlink_to(outside) + app = self.app() + self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) + self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) + + def test_create_applies_initial_overlay_after_seed_and_starts_without_cache(self): + self.create() + self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"]) + self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists()) + manifest, contents, _, _ = self.prepare() + payload = self.agent.spec(fixtures.INSTANCE, 1) + payload["port"] += 1 + payload["files"].append({"path": "runtime/plutonium/" + BOOTSTRAPPER, "contentBase64": base64.b64encode(b"custom-overlay").decode()}) + other_id = "00000000-0000-4000-8000-000000000002" + result = self.agent.execute({**self.operation("instance.create", payload, "second-create"), "instanceId": other_id}) + self.assertEqual(result["state"], "completed", result) + app = self.agent.instance_dir(other_id, 1) / "app" + self.assertEqual((app / "runtime/plutonium" / BOOTSTRAPPER).read_bytes(), b"custom-overlay") + self.assertEqual(json.loads((app / "runtime/plutonium/cdn_info.json").read_text()), manifest) + self.assertEqual((app / "configs/server.cfg").read_bytes(), b"original") + + def test_rebuild_seeds_the_new_generation_from_current_cache(self): + payload = self.create() + manifest, contents, _, _ = self.prepare() + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2}, "rebuild")) + self.assertEqual(result["state"], "completed", result) + self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 2)["running"]) + self.assert_runtime(self.agent.instance_dir(fixtures.INSTANCE, 2) / "app", manifest, contents) + + def test_restart_preserves_the_instances_downloaded_runtime(self): + self.prepare() + self.create() + runtime_file = self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium" / BOOTSTRAPPER + runtime_file.write_bytes(b"container-updated-runtime") + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + with patch.object(self.agent, "seed_plutonium", side_effect=AssertionError("Restart must retain its runtime")): + result = self.agent.execute(self.operation("instance.restart", op_id="restart")) + self.assertEqual(result["state"], "completed", result) + self.assertEqual(runtime_file.read_bytes(), b"container-updated-runtime") + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/guides/native-hosts.md b/docs/guides/native-hosts.md index 8114007..9ae2241 100644 --- a/docs/guides/native-hosts.md +++ b/docs/guides/native-hosts.md @@ -55,8 +55,10 @@ enrollment; inspect `systemctl status docker` and `journalctl -u docker`. The controller reconciles automatic setup on agent heartbeats. Command polls only deliver work already queued. Setup first pulls `PLUTAINER_IMAGE` (default `ghcr.io/ayymoss/plutainer:latest`), then prepares the latest verified upload for -T5 and for T6 when available, one setup operation at a time. Uploading new files -or enrolling another host needs no manual propagation. Setup waits for an active, connected, +T5 and for T6 when available, one setup operation at a time. After those prerequisites, +capable agents also prepare the optional Plutonium runtime cache described below. +Uploading new files or enrolling another host needs no manual propagation. +Setup waits for an active, connected, healthy host with sufficient storage. Failed operations retry automatically with backoff from one minute to fifteen minutes; their errors remain visible in the host's setup panel and **Hosts → Operation history**. This shared history lists @@ -90,6 +92,50 @@ cached mutable image tag. Advanced controls on the host and game-file pages can explicitly pull a custom runtime, prepare an older upload or retry preparation immediately. +### Plutonium runtime cache + +Agent 1.8.0 adds a host cache for the files Plutainer normally downloads from +Plutonium on a new container's first start. This is separate from the cached +Docker image and uploaded base game libraries. Automatic preparation warms it +after the image and libraries are ready and refreshes it for new production +revisions. The controller only queues this work for agents that advertise support. +Cache preparation is optional: a missing or corrupt cache does not prevent a +launch from using Plutainer's normal download path. + +Cache preparation has lower priority than deployments and other host changes. +If work arrives during a download, the next heartbeat tells the agent to yield +between transfer steps. Preparation resumes under the same operation ID after +that work finishes, reusing files that have already passed verification. + +The agent downloads the official production manifest and its listed vendor files, +verifies the SHA-1 hashes published in that manifest, and makes a snapshot +available only after the complete set passes verification. It keeps the cache +under `agent/plutonium-cache` within the host data directory. It does not populate +the cache from existing lobbies, so lobby configuration, keys, custom scripts, +plugins and runtime state are not collected into it. Preparation retains the +current snapshot and the resumable download, removes older snapshots, and clears +abandoned temporary files before checking available storage or publishing the +new snapshot. A failed refresh preserves the previous complete cache. + +For new instances and rebuilds using `ghcr.io/ayymoss/plutainer` image references, +the agent seeds `app/runtime/plutonium` from a valid snapshot, including its +matching `cdn_info.json`, before applying the instance's launch files. Each +instance receives its own writable copy; changes made by a lobby or its updater +cannot change the host cache or another lobby. Custom runtime images keep their +existing startup behavior. + +Plutainer's updater remains enabled. When a cached revision is current, it can +skip the initial vendor-file download; when production is newer, it can download +the update normally. Launch readiness still verifies the running Plutonium +revision against production. Refreshing the cache does not restart or update +running lobbies. Ordinary restarts retain the instance's existing files and do +not need to seed from the cache again. + +Deleting a lobby preserves the host cache for subsequent launches. Deleting the +host removes it with the rest of the managed data directory. This is intended to +reduce repeated first-start downloads; deployment-time improvements still need +measurement on a real Debian/Docker/Plutonium host. + ### Storage layout The default host root is `/var/lib/electric-cherry`; installer `--data-dir` changes @@ -698,6 +744,15 @@ game hosts. Run: loss and host reboot during update. Verify updater protocol mismatch requests an installer rerun. - T5 and T6 deployments and a mixed collection from a fresh destination host. +- Warm the Plutonium runtime cache, then create and rebuild T5 and T6 lobbies; + verify the running revision and compare first-start downloads and launch times + with an empty cache. Test missing, corrupt and interrupted cache preparation, + a newer production revision, and a custom runtime image. Confirm launch-file + overrides survive seeding, lobby writes cannot alter the cache or another lobby, + and cache refresh leaves running lobbies unchanged. Queue a deployment during + preparation and verify that downloading yields and later resumes. Interrupt + preparation by stopping the agent and verify that retry removes abandoned + temporary files while retaining completed downloads. - Start, stop, restart, rebuild, destroy; verify RCON, map and runtime revision. - Read console and game logs, including Plutainer's relative game-log links. - File campaign and rollback, with a player joining between controller check and diff --git a/src/lib/hosts.ts b/src/lib/hosts.ts index af77af1..c4e0ef1 100644 --- a/src/lib/hosts.ts +++ b/src/lib/hosts.ts @@ -82,6 +82,7 @@ export interface HostInventory { directoriesWritable?: boolean; runtimeImages?: string[]; libraries?: HostLibrary[]; currentOperation?: string | null; [key: string]: unknown; capabilities?: string[]; + plutoniumCache?: { revision: number; bytes: number }; agentUpdate?: HostAgentUpdate; agentLanes?: HostAgentLane[]; instanceResources?: InstanceResources[]; @@ -146,7 +147,7 @@ export interface HostOperationHistoryPage { } export interface HostPreparationItem { - kind: 'runtime' | 'library'; label: string; + kind: 'runtime' | 'library' | 'plutonium'; label: string; state: 'waiting' | 'preparing' | 'ready' | 'blocked'; detail: string; operationId?: string; bytesDownloaded?: number; bytesTotal?: number; cache?: HostLibraryCache; diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index b6b437e..ccdea53 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -277,6 +277,12 @@ export function validateInventory(value: unknown): HostInventory { for (const key of ['dockerRunning','directoriesWritable']) if (inventory[key] !== undefined && typeof inventory[key] !== 'boolean') throw new Error(`Invalid inventory ${key}.`); if (inventory.runtimeImages !== undefined && (!Array.isArray(inventory.runtimeImages) || inventory.runtimeImages.some(image => typeof image !== 'string'))) throw new Error('Invalid runtime image inventory.'); if (inventory.capabilities !== undefined && (!Array.isArray(inventory.capabilities) || inventory.capabilities.length > 32 || inventory.capabilities.some(value => typeof value !== 'string' || value.length > 80))) throw new Error('Invalid agent capability inventory.'); + if (inventory.plutoniumCache !== undefined) { + const cache = inventory.plutoniumCache; + if (!cache || typeof cache !== 'object' || Array.isArray(cache) || !Number.isSafeInteger(cache.revision) || cache.revision < 1 + || !Number.isSafeInteger(cache.bytes) || cache.bytes < 0) throw new Error('Invalid Plutonium cache inventory.'); + inventory.plutoniumCache = { revision: cache.revision, bytes: cache.bytes }; + } if (inventory.agentUpdate !== undefined) { const update = inventory.agentUpdate; const version = (value: unknown) => typeof value === 'string' && value.length <= 80 && /^\d+\.\d+\.\d+$/.test(value); @@ -396,6 +402,14 @@ export type OperationDeliveryRow = Omit; export type HostOperationStatus = Omit; const operationStatusColumns = 'o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.result,o.error,o.created_at,o.updated_at'; const operationPayloadSql = 'COALESCE((SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload)'; +const readOnlyOperationSql = `(o.kind IN ('host.inspect','instance.inspect','rcon.status-batch','files.read','files.list','files.stat','files.manifest','logs.read') + OR (o.kind='rcon.execute' AND COALESCE(json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version'),0)))`; +/** Optional cache downloads yield their write lane to deployments and host changes. */ +export function shouldDeferPlutoniumPreparation(hostId: string): boolean { + return !!database().prepare(`SELECT 1 FROM ec_host_operations o WHERE o.host_id=? + AND o.state IN ('pending','accepted','running','unknown') AND o.kind!='plutonium.prepare' + AND NOT ${readOnlyOperationSql} LIMIT 1`).get(hostId); +} function operationStatusFromRow(row: Omit): HostOperationStatus { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; } function operationFromRow(row: OperationRow): HostOperation { const status = operationStatusFromRow(row); return { ...status, result: unpackFileJson(status.result), payload: unpackFileJson(JSON.parse(row.payload)) }; } export function getHostOperation(id: string): HostOperation | null { @@ -483,7 +497,7 @@ function enqueueHostOperation(hostId: string, kind: string, payload: string | Bu } } if (['instance.create','create'].includes(kind)) requireReadyHost(hostId); - if (host.mode === 'disabled' && kind !== 'instance.cancel-create' && !(host.enablingSince && ['runtime.pull','library.download','host.inspect'].includes(kind))) throw new Error('Host is disabled. Enable it before requesting operations.'); + if (host.mode === 'disabled' && kind !== 'instance.cancel-create' && !(host.enablingSince && ['runtime.pull','library.download','plutonium.prepare','host.inspect'].includes(kind))) throw new Error('Host is disabled. Enable it before requesting operations.'); const requestHash = createHash('sha256').update(payload).digest('hex'); const packed = packFileJson(JSON.parse(payload.toString())); payload = JSON.stringify(packed.value); @@ -523,12 +537,10 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return []; // Accepted/running/unknown commands keep the same immutable ID. The agent // consults its durable journal before acknowledging or executing a replay. - const readOnly = `(kind IN ('host.inspect','instance.inspect','rcon.status-batch','files.read','files.list','files.stat','files.manifest','logs.read') - OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`; const healthOnly = `(kind IN ('instance.inspect','rcon.status-batch') OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`; const healthLane = host.inventory.capabilities?.includes('operations.health-lane'); const laneFilter = lane === 'all' ? '' : lane === 'health' ? ` AND ${healthOnly}` - : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}${lane === 'read' && healthLane ? ` AND NOT ${healthOnly}` : ''}`; + : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnlyOperationSql}${lane === 'read' && healthLane ? ` AND NOT ${healthOnly}` : ''}`; // Health never queues behind a delivered file batch. Keep background reads // in small batches to amortize delivery without handing out a long backlog. const limit = healthLane ? lane === 'health' ? 1 : lane === 'read' ? 4 : 25 : 25; @@ -536,13 +548,14 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' const columns = includePayload ? `${operationStatusColumns},${operationPayloadSql} AS payload` : operationStatusColumns; const rows = database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown') ${laneFilter} - ${host.mode === 'disabled' ? "AND kind IN ('runtime.pull','library.download','host.inspect')" : ''} + ${host.mode === 'disabled' ? "AND kind IN ('runtime.pull','library.download','plutonium.prepare','host.inspect')" : ''} AND (? OR state != 'pending' OR kind NOT IN ('instance.create','create')) AND (kind!='instance.cancel-create' OR ?) AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id) ${lane === 'health' ? "AND (state NOT IN ('accepted','running') OR MAX(updated_at,COALESCE(last_delivered_at,'')) <= ?)" : ''} - ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`) + ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 WHEN kind='plutonium.prepare' THEN 2 ELSE 1 END, + COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`) // A short replay lease avoids handing the idle health lane its busy // sibling's work, but still recovers a lost receipt or agent restart. .all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false), diff --git a/src/lib/server/hosts/observations.ts b/src/lib/server/hosts/observations.ts index 1e877fa..5398b7c 100644 --- a/src/lib/server/hosts/observations.ts +++ b/src/lib/server/hosts/observations.ts @@ -35,6 +35,10 @@ export const importantOperationSql = `(o.routine=0 OR o.state IN ('failed','unkn const payloadSql = 'COALESCE((SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload)'; /** Deliberately exclude arbitrary RCON commands and all file contents/credentials. */ export const operationTargetSql = `CASE + WHEN o.kind='plutonium.prepare' THEN CASE + WHEN json_type(${payloadSql},'$.revision')='integer' AND json_extract(${payloadSql},'$.revision')>0 + THEN 'Production revision r' || json_extract(${payloadSql},'$.revision') + ELSE 'Latest production revision' END WHEN o.kind IN ('files.read','files.stat','files.list','logs.read') THEN CASE WHEN json_type(${payloadSql},'$.path')='text' THEN substr(json_extract(${payloadSql},'$.path'),1,1024) END WHEN o.kind='files.manifest' THEN substr((SELECT group_concat(value,', ') FROM json_each(${payloadSql},'$.roots') WHERE type='text'),1,1024) diff --git a/src/lib/server/hosts/preparation.ts b/src/lib/server/hosts/preparation.ts index 313889d..923cd2f 100644 --- a/src/lib/server/hosts/preparation.ts +++ b/src/lib/server/hosts/preparation.ts @@ -1,15 +1,17 @@ import { db } from '../db/index.js'; import { config } from '../config.js'; -import { AGENT_PROTOCOL_VERSION, getHost, queueHostOperation } from './index.js'; +import { AGENT_PROTOCOL_VERSION, getHost, queueHostOperation, shouldDeferPlutoniumPreparation } from './index.js'; import { getHostLibraryCache, latestPublishedLibraries, queueLibraryDownload } from '../game-files/index.js'; import { GAME_FILE_GAMES } from '../../game-files.js'; import type { Host, HostOperationState, HostPreparation, HostPreparationItem } from '../../hosts.js'; import { scheduleAvailabilityRefresh } from './availability.js'; +import { readSetting } from '../maintenance/settings.js'; +import type { PlutoniumRelease } from '../../plutonium.js'; const activeStates = ['pending', 'accepted', 'running', 'unknown']; const supersededDownload = 'Superseded by newer game files before delivery to the host.'; type PreparationOperation = { id: string; sequence: number; last_delivered_at: string | null; state: HostOperationState; updated_at: string; error: string | null }; -type Target = { item: HostPreparationItem; image?: string; versionId?: string; requiredBytes: number; operation?: PreparationOperation }; +type Target = { item: HostPreparationItem; image?: string; versionId?: string; revision?: number; requiredBytes: number; operation?: PreparationOperation }; type QueuedSelection = { libraryId: string }; type LibraryTarget = Pick[number], 'id' | 'sizeBytes' | 'manifestSha256'> & { name: string; queued?: boolean }; @@ -48,25 +50,27 @@ function targetLibraries(selections: QueuedSelection[]): LibraryTarget[] { return [...queued.values(), ...latest]; } -function latestOperation(hostId: string, kind: string, field: string, value: string): PreparationOperation | undefined { +function latestOperation(hostId: string, kind: string, field?: string, value?: string): PreparationOperation | undefined { return db().prepare(`SELECT id,rowid AS sequence,last_delivered_at,state,updated_at,error FROM ec_host_operations - WHERE host_id=? AND kind=? AND json_extract(payload,?)=? ORDER BY created_at DESC,rowid DESC LIMIT 1`) - .get(hostId, kind, `$.${field}`, value) as PreparationOperation | undefined; + WHERE host_id=? AND kind=? ${field ? 'AND json_extract(payload,?)=?' : ''} ORDER BY created_at DESC,rowid DESC LIMIT 1`) + .get(hostId, kind, ...(field ? [`$.${field}`, value!] : [])) as PreparationOperation | undefined; } function retryAt(hostId: string, target: Target): number { const operation = target.operation; if (!operation || operation.state !== 'failed') return 0; - const kind = target.image ? 'runtime.pull' : 'library.download'; - const field = target.image ? '$.image' : '$.libraryVersionId'; - const value = target.image ?? target.versionId!; + const kind = target.image ? 'runtime.pull' : target.versionId ? 'library.download' : 'plutonium.prepare'; + const field = target.image ? '$.image' : target.versionId ? '$.libraryVersionId' : null; + const value = target.image ?? target.versionId; + const match = field ? 'AND json_extract(payload,?)=?' : ''; + const bindings = [hostId, kind, ...(field ? [field, value!] : [])]; // Count consecutive failures, including across controller restarts. Unknown // operations are never retried under a new ID: their journal must reconcile. const { count } = db().prepare(`SELECT COUNT(*) AS count FROM ec_host_operations - WHERE host_id=? AND kind=? AND json_extract(payload,?)=? AND state='failed' + WHERE host_id=? AND kind=? ${match} AND state='failed' AND rowid>COALESCE((SELECT MAX(rowid) FROM ec_host_operations - WHERE host_id=? AND kind=? AND json_extract(payload,?)=? AND state='completed'),0)`) - .get(hostId, kind, field, value, hostId, kind, field, value) as { count: number }; + WHERE host_id=? AND kind=? ${match} AND state='completed'),0)`) + .get(...bindings, ...bindings) as { count: number }; return Date.parse(operation.updated_at) + Math.min(15 * 60_000, 60_000 * 2 ** Math.min(4, Math.max(0, count - 1))); } @@ -108,6 +112,22 @@ function targets(host: Host): Target[] { : ready ? 'The latest game files are downloaded and verified.' : 'The latest game files will be downloaded and verified automatically.', bytesDownloaded: cache?.bytesDownloaded ?? 0, bytesTotal: version.sizeBytes, ...(cache ? { cache } : {}) } }); } + if (host.inventory.capabilities?.includes('plutonium.cache')) { + // Read the same production release evidence as freshRelease without + // importing its fleet/provisioning observers into host preparation. + const release = readSetting('plutonium_release', { revision: null, checkedAt: null, error: null }); + const age = release.checkedAt ? Date.now() - Date.parse(release.checkedAt) : NaN; + const revision = !release.error && age >= 0 && age < 60 * 60_000 + && Number.isSafeInteger(release.revision) && release.revision! > 0 ? release.revision! : 0; + const cache = host.inventory.plutoniumCache; + const ready = !!cache && cache.revision >= revision; + result.push({ revision, requiredBytes: host.minimumFreeBytes, + operation: latestOperation(host.id, 'plutonium.prepare'), + item: { kind: 'plutonium', label: 'Plutonium cache (optional)', state: ready ? 'ready' : 'waiting', + detail: ready ? `Production revision r${cache.revision} is cached for new deployments.` + : cache ? `Production revision r${revision} will replace cached r${cache.revision} automatically.` + : 'Plutonium files will be cached after the runtime and game files are ready.' } }); + } const blocker = hostBlocker(host); for (const target of result) { if (target.item.state === 'ready') continue; @@ -116,14 +136,15 @@ function targets(host: Host): Target[] { if (blocker) { Object.assign(target.item, blocker); continue; } if (operation && activeStates.includes(operation.state)) { target.item.state = 'preparing'; - target.item.detail = !target.image && operation.state === 'pending' && !operation.last_delivered_at + target.item.detail = target.versionId && operation.state === 'pending' && !operation.last_delivered_at ? 'Queued for game-file preparation; waiting for agent delivery or fleet preparation capacity.' : operation.state === 'unknown' ? 'Waiting for the agent to reconcile interrupted preparation.' : target.image ? 'Downloading the runtime automatically.' + : target.revision !== undefined ? 'Downloading and verifying the optional Plutonium cache.' : target.item.cache?.status === 'verifying' ? 'Download complete. Verifying all game files.' : 'Downloading game files. Verification starts after all files arrive.'; } else if (operation && Date.parse(host.lastHeartbeat!) <= Date.parse(operation.updated_at)) { target.item.state = 'preparing'; target.item.detail = 'Waiting for a fresh host inventory after preparation.'; - } else if ((!enabling || (operation?.sequence ?? 0) > enabling.operation_floor) && retryAt(host.id, target) > Date.now()) { + } else if ((target.revision !== undefined || !enabling || (operation?.sequence ?? 0) > enabling.operation_floor) && retryAt(host.id, target) > Date.now()) { target.item.state = 'blocked'; target.item.detail = `${operation?.error ?? 'Preparation failed.'} Automatic retry after ${new Date(retryAt(host.id, target)).toISOString()}.`; } else if (Number(host.inventory.storageAvailable ?? 0) < target.requiredBytes) { @@ -142,16 +163,19 @@ export function hostPreparationStatus(host: Host): HostPreparation { } } else if (!items.some(item => item.kind === 'library')) items.push({ kind: 'library', label: 'Game files', state: 'waiting', detail: 'Upload T5 or T6 game files; hosts download the latest available files automatically.' }); - const state = items.some(item => item.state === 'blocked') ? 'blocked' - : items.some(item => item.state === 'preparing') ? 'preparing' - : items.some(item => item.state === 'waiting') ? 'waiting' : 'ready'; + // Cache health is visible, but never makes a usable host ineligible or + // delays re-enabling: a lobby's normal updater handles cache misses. + const required = items.filter(item => item.kind !== 'plutonium'); + const state = required.some(item => item.state === 'blocked') ? 'blocked' + : required.some(item => item.state === 'preparing') ? 'preparing' + : required.some(item => item.state === 'waiting') ? 'waiting' : 'ready'; return { state, items }; } export function presentHostPreparation(host: Host): Host { const preparation = hostPreparationStatus(host); if (['needs-runtime', 'needs-game-files'].includes(host.readiness) && preparation.state === 'preparing') { - return { ...host, readiness: 'preparing', readinessReasons: preparation.items.filter(item => item.state !== 'ready').map(item => item.detail), preparation }; + return { ...host, readiness: 'preparing', readinessReasons: preparation.items.filter(item => item.kind !== 'plutonium' && item.state !== 'ready').map(item => item.detail), preparation }; } return { ...host, preparation }; } @@ -180,20 +204,26 @@ export function reconcileHostPreparation(hostId: string): HostPreparation { if (!host) throw new Error('Host was not found.'); if (!hostBlocker(host)) { supersedeUndeliveredDownloads(hostId); - // Queue only one setup mutation at a time so each capacity check uses - // fresh disk inventory. This also respects manually queued downloads. + // Queue one required setup mutation at a time with fresh disk + // inventory. A running optional cache yields to required work via + // the heartbeat flag, then resumes under its original operation ID. const active = db().prepare(`SELECT 1 FROM ec_host_operations WHERE host_id=? AND kind IN ('runtime.pull','library.download','library.import') AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId); const finished = db().prepare(`SELECT updated_at FROM ec_host_operations WHERE host_id=? - AND kind IN ('runtime.pull','library.download','library.import') AND state IN ('completed','failed') + AND kind IN ('runtime.pull','library.download','library.import','plutonium.prepare') AND state IN ('completed','failed') AND (error IS NULL OR error != ?) ORDER BY updated_at DESC LIMIT 1`).get(hostId, supersededDownload) as { updated_at: string } | undefined; if (!active && (!finished || Date.parse(host.lastHeartbeat!) > Date.parse(finished.updated_at))) { // A failed or oversized library must not prevent other libraries // from preparing. The runtime still comes first when eligible. - const target = targets(host).find(target => target.item.state === 'waiting'); + const candidates = targets(host); + const requiredReady = candidates.every(target => target.item.kind === 'plutonium' || target.item.state === 'ready'); + const target = candidates.find(target => target.item.state === 'waiting' && (target.item.kind !== 'plutonium' || requiredReady)); if (target?.image) queueHostOperation(host.id, 'runtime.pull', { image: target.image }); else if (target?.versionId) queueLibraryDownload(host.id, target.versionId, { automatic: true }); + else if (target?.revision !== undefined && !shouldDeferPlutoniumPreparation(host.id)) queueHostOperation(host.id, 'plutonium.prepare', { + revision: target.revision, minimumFreeBytes: host.minimumFreeBytes, automatic: true + }); } } const status = hostPreparationStatus(host); diff --git a/src/lib/ui/components/HostPreparation.svelte b/src/lib/ui/components/HostPreparation.svelte index 5e77293..1d93c6a 100644 --- a/src/lib/ui/components/HostPreparation.svelte +++ b/src/lib/ui/components/HostPreparation.svelte @@ -7,6 +7,7 @@

The default runtime and the latest available T5 and T6 game files are prepared automatically. Re-enabling a host requires a fresh runtime pull and verification of both BO1 (T5) and BO2 (T6) libraries before it accepts deployments. Running servers keep their current files.

+

Supported hosts also cache Plutonium files to speed up new deployments. This optional cache refreshes when a new production release is detected; deployments can proceed while it is unavailable.

{#if host.preparation}

Automatic setup: {host.preparation.state === 'ready' ? 'Ready' : host.preparation.state === 'preparing' ? 'In progress' : host.preparation.state === 'blocked' ? 'Needs attention' : 'Waiting'}

    diff --git a/src/routes/api/agent/heartbeat/+server.ts b/src/routes/api/agent/heartbeat/+server.ts index dc5773b..b452dc7 100644 --- a/src/routes/api/agent/heartbeat/+server.ts +++ b/src/routes/api/agent/heartbeat/+server.ts @@ -1,5 +1,5 @@ import { json } from '@sveltejs/kit'; -import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index.js'; +import { heartbeat, pendingCreationCancellations, shouldDeferPlutoniumPreparation } from '$lib/server/hosts/index.js'; import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; import { reconcileHostLibraryCleanup } from '$lib/server/game-files/index.js'; @@ -17,6 +17,7 @@ export async function POST({ request }: { request: Request }) { reconcileHostPreparation(host.id); reconcileHostLibraryCleanup(host.id); return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15, + deferPlutoniumPreparation: shouldDeferPlutoniumPreparation(host.id), observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() }); } catch (cause) { return hostApiError(cause); } } diff --git a/src/routes/hosts/operations/+page.svelte b/src/routes/hosts/operations/+page.svelte index c5ed114..ebdaad5 100644 --- a/src/routes/hosts/operations/+page.svelte +++ b/src/routes/hosts/operations/+page.svelte @@ -90,7 +90,7 @@ OperationHostTargetSourceStateUpdatedResult {#each history.operations as operation (operation.id)} - {operation.kind}{operation.id} + {operation.kind === 'plutonium.prepare' ? 'Prepare Plutonium cache' : operation.kind}{operation.id} {#if operation.hostName !== null}{operation.hostName}{:else}Deleted host{operation.hostId}{/if} {operation.serverName ?? operation.instanceId ?? 'Host-level'} diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 502d503..01cb0df 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.7.0" +VERSION = "1.8.0" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -42,6 +42,10 @@ MANIFEST_MAX_SECONDS = 15 MANIFEST_MAX_DEPTH = 64 LIBRARY_DOWNLOAD_WORKERS = 4 HEALTH_LOCK_TIMEOUT_SECONDS = 5 +PLUTONIUM_MANIFEST_URL = "https://cdn.plutoniummod.com/updater/prod/info.json" +PLUTONIUM_MANIFEST_LIMIT = 2 * 1024 * 1024 +PLUTONIUM_FILE_LIMIT = 512 * 1024 * 1024 +PLUTONIUM_CACHE_LIMIT = 4 * 1024 * 1024 * 1024 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") AGENT_CONFIG = Path("/etc/electric-cherry/agent.json") AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py") @@ -487,6 +491,34 @@ def managed_write(root, value, content, uid=1000, gid=1000): os.unlink(temp_name, dir_fd=parent) +def managed_copy(root, value, source, uid=1000, gid=1000): + """Copy a private file, using Linux reflinks when supported, never hardlinks.""" + with managed_open(Path(source).parent, Path(source).name) as incoming: + with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name): + temp_name = ".ec-" + uuid.uuid4().hex + fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent) + try: + os.fchmod(fd, 0o644) + if os.geteuid() == 0: + os.fchown(fd, uid, gid) + try: + if not sys.platform.startswith("linux"): + raise OSError("Reflinks unavailable") + fcntl.ioctl(fd, 0x40049409, incoming.fileno()) # FICLONE + except OSError: + os.ftruncate(fd, 0) + with os.fdopen(fd, "wb", closefd=False) as outgoing: + shutil.copyfileobj(incoming, outgoing, 1024 * 1024) + outgoing.flush() + os.fsync(fd) + os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent) + os.fsync(parent) + finally: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + os.unlink(temp_name, dir_fd=parent) + + class ReadWriteGate: """Let observations run together; global mutations wait for existing readers.""" def __init__(self): @@ -921,11 +953,13 @@ class Agent: self.read_claims = set() self.health_probe_gate = threading.BoundedSemaphore(2) self.cancelled_creations = frozenset() + self.defer_plutonium_preparation = False self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.library_progress = {} self.library_transfers = {} self.inventory_manifest_cache = {} self.inventory_manifest_lock = threading.Lock() + self.plutonium_lock = threading.RLock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None self.lane_lock = threading.Lock() @@ -1360,8 +1394,9 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) - return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} + capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) + plutonium_cache = self.plutonium_cache_inventory() + return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} def safe_to_disrupt(self, instance_id, generation): health = self.inspect(instance_id, generation) @@ -1510,6 +1545,223 @@ class Agent: raise Rejected("Initial files exceed the 64 MiB operation limit") return game, library, image, port, bind, env + @staticmethod + def plutonium_manifest(data): + """Validate the official updater format before using any paths or URLs.""" + if len(data) > PLUTONIUM_MANIFEST_LIMIT: + raise AgentError("Plutonium manifest is too large") + try: + manifest = json.loads(data) + except (ValueError, UnicodeError): + raise AgentError("Invalid Plutonium manifest") from None + if not isinstance(manifest, dict) or manifest.get("product") != "plutonium-core-prod": + raise AgentError("Invalid Plutonium production manifest") + revision = manifest.get("revision") + if type(revision) is not int or not 1 <= revision <= 2147483647: + raise AgentError("Invalid Plutonium revision") + base = manifest.get("baseUrl") + if not isinstance(base, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/(?:[A-Za-z0-9_-]+/)*", base): + raise AgentError("Plutonium downloads must use the official production CDN") + entries = manifest.get("files") + if not isinstance(entries, list) or not 1 <= len(entries) <= 4096: + raise AgentError("Invalid Plutonium file list") + names, total = set(), 0 + for entry in entries: + if not isinstance(entry, dict): + raise AgentError("Invalid Plutonium file entry") + name, size, checksum = entry.get("name"), entry.get("size"), entry.get("hash") + if not isinstance(name, str) or len(name) > 512 or re.search(r'[\x00-\x1f\x7f\\:*?"<>|]', name): + raise AgentError("Invalid Plutonium file path") + path = PurePosixPath(name) + if not path.parts or path.is_absolute() or str(path) != name or any(part in (".", "..") or part.endswith((".", " ")) for part in path.parts) or name.lower() == "cdn_info.json": + raise AgentError("Invalid Plutonium file path") + if name.lower() in names: + raise AgentError("Duplicate Plutonium file path") + names.add(name.lower()) + if type(size) is not int or not 0 <= size <= PLUTONIUM_FILE_LIMIT: + raise AgentError("Invalid Plutonium file size") + if not isinstance(checksum, str) or not re.fullmatch(r"[a-f0-9]{40}", checksum): + raise AgentError("Invalid Plutonium file hash") + total += size + if total > PLUTONIUM_CACHE_LIMIT: + raise AgentError("Plutonium cache exceeds the size limit") + for name in names: + if any(str(parent) in names for parent in PurePosixPath(name).parents if str(parent) != "."): + raise AgentError("Conflicting Plutonium file paths") + if "bin/plutonium-bootstrapper-win32.exe" not in {entry["name"] for entry in entries}: + raise AgentError("Plutonium manifest is missing the server bootstrapper") + # Match plutonium-updater's default: launcher assets are optional. + return manifest, [entry for entry in entries if not entry["name"].startswith("launcher")] + + @staticmethod + def plutonium_file_matches(root, entry): + try: + with managed_open(root, entry["name"]) as stream: + if os.fstat(stream.fileno()).st_size != entry["size"]: + return False + checksum = hashlib.sha1() + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + checksum.update(chunk) + return checksum.hexdigest() == entry["hash"] + except (OSError, AgentError): + return False + + def plutonium_open(self, url): + if not isinstance(url, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/[A-Za-z0-9_./-]+", url): + raise AgentError("Plutonium downloads must use the official production CDN") + class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, hdrs, newurl): + raise AgentError("Plutonium download redirects are not allowed") + # Never send the controller's host credentials to the public CDN. + return urllib.request.build_opener(NoRedirect()).open(url, timeout=30) + + def plutonium_snapshot(self): + root = scoped(self.root / "agent", "plutonium-cache") + pointer = json.loads(managed_read(root, "current.json") or b"null") + if not isinstance(pointer, dict) or not isinstance(pointer.get("key"), str) or not re.fullmatch(r"[a-f0-9]{64}", pointer["key"]): + raise AgentError("Plutonium cache is not prepared") + content = scoped(root, pointer["key"] + "/content") + raw = managed_read(content, "cdn_info.json") + if raw is None or digest(raw) != pointer["key"]: + raise AgentError("Plutonium cache manifest is invalid") + manifest, files = self.plutonium_manifest(raw) + if pointer.get("revision") != manifest["revision"] or pointer.get("bytes") != sum(entry["size"] for entry in files): + raise AgentError("Plutonium cache metadata is invalid") + # Heartbeat checks are cheap. Every file is hashed before it is seeded. + for entry in files: + info = scoped(content, entry["name"]).lstat() + if not stat_module.S_ISREG(info.st_mode) or info.st_size != entry["size"]: + raise AgentError("Plutonium cache files are incomplete") + return pointer, content, raw, files + + def plutonium_cache_inventory(self): + with self.plutonium_lock: + try: + pointer, _, _, _ = self.plutonium_snapshot() + return {"revision": pointer["revision"], "bytes": pointer["bytes"]} + except (OSError, ValueError, AgentError): + return None + + def prepare_plutonium(self, payload): + target, reserve = payload.get("revision", 0), payload.get("minimumFreeBytes", 0) + if type(target) is not int or not 0 <= target <= 2147483647 or type(reserve) is not int or reserve < 0: + raise Rejected("Invalid Plutonium preparation request") + root = scoped(self.root / "agent", "plutonium-cache") + root.mkdir(mode=0o700, exist_ok=True) + os.chmod(root, 0o700) + deadline = time.monotonic() + 1800 + def active(): + if self.stopping.is_set() or time.monotonic() >= deadline: + raise AgentError("Plutonium cache preparation was interrupted or timed out") + if self.defer_plutonium_preparation: + raise Held("Plutonium cache preparation is waiting for other host operations") + active() + try: + with self.plutonium_open(PLUTONIUM_MANIFEST_URL) as response: + raw = response.read(PLUTONIUM_MANIFEST_LIMIT + 1) + manifest, files = self.plutonium_manifest(raw) + if manifest["revision"] < target: + raise AgentError("Plutonium CDN has not reached the requested revision") + key = digest(raw) + previous = None + with self.plutonium_lock: + with contextlib.suppress(OSError, ValueError, AgentError): + _, previous, _, _ = self.plutonium_snapshot() + if previous is not None and previous.parent.name == key and all(self.plutonium_file_matches(previous, entry) for entry in files): + return {"revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)} + staging = scoped(root, key + ".partial/content") + staging.mkdir(parents=True, exist_ok=True) + expected = {entry["name"] for entry in files} + # SIGKILL/power loss bypasses download/copy finally blocks. Remove + # abandoned temporary files before counting space or publishing. + for directory, subdirs, filenames in os.walk(staging, followlinks=False): + for name in subdirs + filenames: + path = Path(directory) / name + if path.is_symlink() or (name in filenames and path.relative_to(staging).as_posix() not in expected): + path.unlink() + # Keep at most the current snapshot and this resumable preparation. + current_key = previous.parent.name if previous is not None else None + for directory in root.iterdir(): + if re.fullmatch(r"[a-f0-9]{64}(?:\.partial)?", directory.name) and directory.name not in (current_key, key + ".partial"): + if directory.is_symlink(): + directory.unlink() + else: + shutil.rmtree(directory) + missing = sum(entry["size"] for entry in files if not self.plutonium_file_matches(staging, entry)) + if shutil.disk_usage(root).free < reserve + missing + len(raw): + raise AgentError("Insufficient free storage for the Plutonium cache and host reserve") + for entry in files: + active() + if self.plutonium_file_matches(staging, entry): + continue + if previous is not None and self.plutonium_file_matches(previous, entry): + managed_copy(staging, entry["name"], scoped(previous, entry["name"]), os.geteuid(), os.getegid()) + continue + destination = scoped(staging, entry["name"]) + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_name(".download-" + uuid.uuid4().hex) + try: + with self.plutonium_open(manifest["baseUrl"] + entry["hash"]) as incoming, temporary.open("xb") as outgoing: + size, checksum = 0, hashlib.sha1() + while True: + active() + chunk = incoming.read(min(1024 * 1024, entry["size"] - size + 1)) + if not chunk: + break + size += len(chunk) + if size > entry["size"]: + raise AgentError("Plutonium download exceeds its expected size") + checksum.update(chunk) + outgoing.write(chunk) + if size != entry["size"] or checksum.hexdigest() != entry["hash"]: + raise AgentError("Plutonium download failed size or checksum verification") + outgoing.flush() + os.fsync(outgoing.fileno()) + os.replace(temporary, destination) + sync_parent(destination) + finally: + with contextlib.suppress(FileNotFoundError): + temporary.unlink() + # Publish metadata last. Interrupted transfers can never claim ready. + active() + atomic_bytes(staging / "cdn_info.json", raw) + pointer = {"key": key, "revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)} + with self.plutonium_lock: + final = scoped(root, key) + if final.exists(): + # Invalidate before replacing a damaged/same-revision cache. + if current_key == key: + (root / "current.json").unlink(missing_ok=True) + shutil.rmtree(final) + os.replace(staging.parent, final) + sync_parent(final) + atomic_json(root / "current.json", pointer) + if previous is not None and current_key != key: + shutil.rmtree(previous.parent) + return {"revision": pointer["revision"], "bytes": pointer["bytes"]} + except (urllib.error.URLError, TimeoutError, http.client.HTTPException): + raise AgentError("Plutonium CDN download is unavailable; the previous cache is retained") from None + + def seed_plutonium(self, app, image): + if not re.fullmatch(r"ghcr\.io/ayymoss/plutainer(?::[A-Za-z0-9_.-]+)?(?:@sha256:[a-f0-9]{64})?", image): + return False + with self.plutonium_lock: + try: + _, content, raw, files = self.plutonium_snapshot() + if not all(self.plutonium_file_matches(content, entry) for entry in files): + scoped(self.root / "agent", "plutonium-cache/current.json").unlink(missing_ok=True) + raise AgentError("Plutonium cache checksum verification failed") + except (OSError, ValueError, AgentError): + return False + uid, gid = int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)) + marker = "runtime/plutonium/cdn_info.json" + managed_write(app, marker, None, uid, gid) + for entry in files: + self.check_creation_cancellation() + managed_copy(app, "runtime/plutonium/" + entry["name"], scoped(content, entry["name"]), uid, gid) + managed_write(app, marker, raw, uid, gid) + return True + def create(self, instance_id, generation, payload): self.check_creation_cancellation() directory = self.instance_dir(instance_id, generation) @@ -1527,6 +1779,7 @@ class Agent: app = directory / "app" app.mkdir(parents=True, exist_ok=True) self.make_writable(app, directory=True) + self.seed_plutonium(app, image) self.write_initial_files(app, payload.get("files", [])) atomic_json(directory / "spec.json", payload) env_path = directory / "environment" @@ -2184,6 +2437,8 @@ class Agent: return {"image": image, "ready": True} if kind == "host.inspect": return self.inventory() + if kind == "plutonium.prepare": + return self.prepare_plutonium(payload) if kind == "rcon.status-batch": return self.rcon_status_batch(operation) if kind == "library.download": @@ -2290,7 +2545,7 @@ class Agent: if not self.config.get("pendingCredential") or self.recover_credential(): return self.journal.update(operation["id"], "completed", {"rotated": True, "reconciled": True}) return row - if kind == "runtime.pull": + if kind in ("runtime.pull", "plutonium.prepare"): # Docker resumes already downloaded layers. Pull again under # the same operation ID: an existing tag can still refer to an # older image and cannot prove an interrupted pull completed. @@ -2298,6 +2553,8 @@ class Agent: try: result = self.dispatch(operation) return self.journal.update(operation["id"], "completed", {**result, "reconciled": True}) + except Held as err: + return self.journal.update(operation["id"], "accepted", {"blocked": "host-busy", "message": str(err)}) except Exception: return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed") if kind == "library.import": @@ -2407,6 +2664,7 @@ class Agent: # Only the heartbeat thread replaces this immutable snapshot; it never # mutates the operation journal or races Docker cleanup against create. self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str)) + self.defer_plutonium_preparation = response.get("deferPlutoniumPreparation") is True keep_days = response.get("observationKeepDays") self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0 host = response.get("host", {}) diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 2d6d71a..176a290 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.7.0", + "version": "1.8.0", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "4b40206f477dcf35f0506281b3dddca32b57214a964bcf5169c8611dc705132d", + "sha256": "eb36afb3113b92304da42fc78ac7f3b8218456f33c5bc1c0f57b018bbc2ad65d", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/host-preparation.test.ts b/tests/unit/host-preparation.test.ts index c3939e1..a7c5902 100644 --- a/tests/unit/host-preparation.test.ts +++ b/tests/unit/host-preparation.test.ts @@ -47,7 +47,7 @@ function heartbeat(id: string, overrides: HostInventory = {}) { } function operations(id: string) { - return hosts.listHostOperations(id).filter(operation => ['runtime.pull', 'library.download'].includes(operation.kind)); + return hosts.listHostOperations(id).filter(operation => ['runtime.pull', 'library.download', 'plutonium.prepare'].includes(operation.kind)); } async function publish(game: 't5' | 't6' = 't5', bytes = content) { @@ -85,10 +85,11 @@ describe('automatic host preparation', () => { } finally { cleanup.mockRestore(); } }); - it('keeps re-enabled hosts disabled until a fresh runtime pull and both game libraries are verified', async () => { + it.each([false, true])('keeps re-enabled hosts disabled until required files are verified, without waiting for the optional cache (supported: %s)', async supported => { const t5 = await publish('t5'); const t6 = await publish('t6'); - const readyInventory = { runtimeImages: [runtime], libraries: [t5, t6].map(version => ({ id: version.id, state: 'ready' as const })) }; + const readyInventory = { runtimeImages: [runtime], libraries: [t5, t6].map(version => ({ id: version.id, state: 'ready' as const })), + capabilities: [...inventory.capabilities!, ...(supported ? ['plutonium.cache'] : [])] }; const id = enroll(readyInventory); verified(id, t5); verified(id, t6); const oldRuntime = hosts.queueHostOperation(id, 'runtime.pull', { image: runtime }); @@ -126,6 +127,7 @@ describe('automatic host preparation', () => { preparation.reconcileHostPreparation(id); expect(hosts.getHost(id)).toMatchObject({ mode: 'active', readiness: 'ready', enablingSince: null }); expect(operations(id).filter(operation => operation.kind === 'library.download')).toHaveLength(2); + expect(operations(id).filter(operation => operation.kind === 'plutonium.prepare')).toHaveLength(supported ? 1 : 0); }); it('requires both games when enabling even if one library is already cached', async () => { @@ -551,3 +553,247 @@ describe('automatic host preparation', () => { expect(libraries.getHostLibraryCache(id, version.id)?.bytesDownloaded).toBe(80); }); }); + +describe('optional host Plutonium cache', () => { + const capable = { capabilities: ['host.decommission', 'plutonium.cache'] }; + async function release(revision: number, overrides: { checkedAt?: string; error?: string | null } = {}) { + const { writeSetting } = await import('../../src/lib/server/maintenance/settings.js'); + writeSetting('plutonium_release', { revision, checkedAt: new Date().toISOString(), error: null, ...overrides }); + } + + it.each(['pending', 'accepted', 'running', 'unknown'] as const)('waits to warm while a deployment is %s', state => { + const ready = { ...capable, runtimeImages: [runtime], libraries: [{ id: 'game-files', state: 'ready' as const }] }; + const id = enroll(ready); + const deployment = hosts.queueHostOperation(id, 'instance.create', {}); + if (state !== 'pending') hosts.reportHostOperation(id, deployment.id, { state }); + preparation.reconcileHostPreparation(id); + expect(hosts.shouldDeferPlutoniumPreparation(id)).toBe(true); + expect(operations(id)).toEqual([]); + hosts.reportHostOperation(id, deployment.id, { state: 'completed' }); + heartbeat(id, ready); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toMatchObject([{ kind: 'plutonium.prepare' }]); + }); + + it('delivers newer deployments before queued cache work and before a cache replay', () => { + const id = enroll({ ...capable, runtimeImages: [runtime], libraries: [{ id: 'game-files', state: 'ready' }] }); + preparation.reconcileHostPreparation(id); + const cache = operations(id)[0]; + const deployment = hosts.queueHostOperation(id, 'instance.create', {}); + expect(hosts.pendingHostOperations(id, 'write').map(operation => operation.id)).toEqual([deployment.id, cache.id]); + hosts.reportHostOperation(id, cache.id, { state: 'running' }); + const change = hosts.queueHostOperation(id, 'files.stage', {}); + const delivery = hosts.pendingHostOperations(id, 'write').map(operation => operation.id); + expect(delivery.at(-1)).toBe(cache.id); + expect(delivery).toContain(deployment.id); + expect(delivery).toContain(change.id); + }); + + it('signals deferral on heartbeats for mutations while ignoring cache work and all observation lanes', async () => { + const host = hosts.createHost({ name: 'Cache yielding host' }); + const ready = { ...inventory, ...capable, runtimeImages: [runtime], libraries: [{ id: 'game-files', state: 'ready' as const }] }; + const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, ready); + preparation.reconcileHostPreparation(host.id); + const cache = operations(host.id)[0]; + hosts.reportHostOperation(host.id, cache.id, { state: 'running' }); + for (const kind of ['host.inspect', 'instance.inspect', 'rcon.status-batch', 'files.read', 'files.list', 'files.stat', 'files.manifest', 'logs.read']) { + hosts.queueHostOperation(host.id, kind, {}); + } + for (const command of ['status', 'getstatus', 'version']) hosts.queueHostOperation(host.id, 'rcon.execute', { command }); + // Work belonging to another host must not pause this one's cache. + const other = enroll(ready); + hosts.queueHostOperation(other, 'instance.restart', {}); + const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js'); + const poll = async () => { + vi.setSystemTime(Date.now() + 1); + const response = await POST({ request: new Request('https://controller.test/api/agent/heartbeat', { + method: 'POST', headers: { Authorization: `Bearer ${credential}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ inventory: ready }) + }) }); + expect(response.status).toBe(200); + return response.json(); + }; + expect(await poll()).toMatchObject({ deferPlutoniumPreparation: false }); + const change = hosts.queueHostOperation(host.id, 'rcon.execute', { command: 'map_restart' }); + expect(await poll()).toMatchObject({ deferPlutoniumPreparation: true }); + hosts.reportHostOperation(host.id, change.id, { state: 'failed', error: 'Command rejected' }); + expect(await poll()).toMatchObject({ deferPlutoniumPreparation: false }); + expect(operations(host.id).filter(operation => operation.kind === 'plutonium.prepare').map(operation => operation.id)).toEqual([cache.id]); + }); + + it('lets newly published game files interrupt cache refresh, then resumes the same cache operation', async () => { + const version = await publish(); + const ready = { ...capable, runtimeImages: [runtime], libraries: [{ id: version.id, state: 'ready' as const }], + plutoniumCache: { revision: 5354, bytes: 200 } }; + const id = enroll(ready); + verified(id, version); + await release(5355); + preparation.reconcileHostPreparation(id); + const cache = operations(id)[0]; + expect(cache).toMatchObject({ kind: 'plutonium.prepare', payload: { revision: 5355 } }); + hosts.reportHostOperation(id, cache.id, { state: 'running' }); + const latest = await publish(); + heartbeat(id, ready); + preparation.reconcileHostPreparation(id); + const download = operations(id)[0]; + expect(download).toMatchObject({ kind: 'library.download', payload: { libraryVersionId: latest.id } }); + expect(hosts.shouldDeferPlutoniumPreparation(id)).toBe(true); + expect(hosts.pendingHostOperations(id, 'write').map(operation => operation.id)).toEqual([download.id, cache.id]); + verified(id, latest); + hosts.reportHostOperation(id, download.id, { state: 'completed' }); + heartbeat(id, { ...ready, libraries: [{ id: latest.id, state: 'ready' }] }); + preparation.reconcileHostPreparation(id); + expect(hosts.shouldDeferPlutoniumPreparation(id)).toBe(false); + expect(hosts.pendingHostOperations(id, 'write').map(operation => operation.id)).toEqual([cache.id]); + expect(operations(id).filter(operation => operation.kind === 'plutonium.prepare')).toHaveLength(1); + }); + + it('warms only after required files, preserves launch readiness, and waits for fresh cache inventory', async () => { + const version = await publish(); + await release(5354); + const id = enroll(capable); + preparation.reconcileHostPreparation(id); + const pull = operations(id)[0]; + expect(pull.kind).toBe('runtime.pull'); + hosts.reportHostOperation(id, pull.id, { state: 'completed' }); + heartbeat(id, { ...capable, runtimeImages: [runtime] }); + preparation.reconcileHostPreparation(id); + const download = operations(id)[0]; + expect(download.kind).toBe('library.download'); + verified(id, version); + hosts.reportHostOperation(id, download.id, { state: 'completed' }); + const ready = { ...capable, runtimeImages: [runtime], libraries: [{ id: version.id, state: 'ready' as const }] }; + heartbeat(id, ready); + expect(preparation.reconcileHostPreparation(id).state).toBe('ready'); + const cache = operations(id)[0]; + expect(cache).toMatchObject({ kind: 'plutonium.prepare', payload: { + revision: 5354, minimumFreeBytes: hosts.getHost(id)!.minimumFreeBytes, automatic: true + } }); + expect(hosts.requireReadyHost(id).readiness).toBe('ready'); + expect(hosts.pendingHostOperations(id, 'read')).toEqual([]); + expect(hosts.pendingHostOperations(id, 'write').map(operation => operation.id)).toEqual([cache.id]); + expect(hosts.listHostOperationHistory(1, 'important', { host: id }).operations[0]).toMatchObject({ + id: cache.id, routine: false, targetPath: 'Production revision r5354' + }); + preparation.reconcileHostPreparation(id); + hosts.reportHostOperation(id, cache.id, { state: 'completed' }); + const pendingInventory = preparation.reconcileHostPreparation(id); + expect(pendingInventory.items.find(item => item.kind === 'plutonium')).toMatchObject({ state: 'preparing' }); + expect(operations(id)).toHaveLength(3); + heartbeat(id, { ...ready, plutoniumCache: { revision: 5354, bytes: 25_000 } }); + expect(preparation.reconcileHostPreparation(id).items.find(item => item.kind === 'plutonium')).toMatchObject({ + state: 'ready', detail: expect.stringContaining('r5354') + }); + const { closeDb } = await import('../../src/lib/server/db/index.js'); + closeDb(); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(3); + }); + + it('refreshes for newer production releases and reconciles an interrupted refresh without duplicates', async () => { + const ready = { ...capable, runtimeImages: [runtime], plutoniumCache: { revision: 5354, bytes: 25_000 } }; + const id = enroll(ready); + await release(5354); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toEqual([]); + await release(5355); + preparation.reconcileHostPreparation(id); + const cache = operations(id)[0]; + expect(cache.payload.revision).toBe(5355); + hosts.reportHostOperation(id, cache.id, { state: 'unknown' }); + await release(5356); + heartbeat(id, ready); + const { closeDb } = await import('../../src/lib/server/db/index.js'); + closeDb(); + expect(preparation.reconcileHostPreparation(id).items.find(item => item.kind === 'plutonium')).toMatchObject({ + state: 'preparing', detail: expect.stringContaining('reconcile interrupted') + }); + expect(operations(id)).toHaveLength(1); + expect(hosts.pendingHostOperations(id, 'write').map(operation => operation.id)).toEqual([cache.id]); + hosts.reportHostOperation(id, cache.id, { state: 'completed' }); + heartbeat(id, { ...ready, plutoniumCache: { revision: 5356, bytes: 26_000 } }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + await release(5355); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + }); + + it.each(['missing', 'stale', 'failed'] as const)('uses the agent production manifest when release evidence is %s', async evidence => { + if (evidence === 'stale') await release(5354, { checkedAt: new Date(Date.now() - 60 * 60_000).toISOString() }); + if (evidence === 'failed') await release(5354, { error: 'Manifest unavailable' }); + const id = enroll({ ...capable, runtimeImages: [runtime] }); + preparation.reconcileHostPreparation(id); + expect(operations(id)[0]).toMatchObject({ kind: 'plutonium.prepare', payload: { revision: 0 } }); + expect(hosts.listHostOperationHistory(1, 'important', { host: id }).operations[0].targetPath).toBe('Latest production revision'); + const cached = enroll({ ...capable, runtimeImages: [runtime], plutoniumCache: { revision: 5353, bytes: 200 } }); + preparation.reconcileHostPreparation(cached); + expect(operations(cached)).toEqual([]); + }); + + it('leaves agents without the cache capability unchanged', async () => { + await release(5355); + const id = enroll({ runtimeImages: [runtime], plutoniumCache: { revision: 5354, bytes: 200 } }); + expect(preparation.reconcileHostPreparation(id).items.some(item => item.kind === 'plutonium')).toBe(false); + expect(operations(id)).toEqual([]); + }); + + it('backs off failed cache work without blocking launch readiness or newly published game files', async () => { + const version = await publish(); + const ready = { ...capable, runtimeImages: [runtime], libraries: [{ id: version.id, state: 'ready' as const }] }; + const id = enroll(ready); + verified(id, version); + await release(5354); + preparation.reconcileHostPreparation(id); + const cache = operations(id)[0]; + hosts.reportHostOperation(id, cache.id, { state: 'failed', error: 'Plutonium CDN unavailable' }); + const failedAt = Date.now(); + heartbeat(id, ready); + const status = preparation.reconcileHostPreparation(id); + expect(status).toMatchObject({ state: 'ready' }); + expect(status.items.find(item => item.kind === 'plutonium')).toMatchObject({ + state: 'blocked', detail: expect.stringContaining('Automatic retry') + }); + expect(hosts.requireReadyHost(id).readiness).toBe('ready'); + expect(operations(id)).toHaveLength(1); + // A changed or expired release must not bypass a failed cache's backoff. + await release(5355); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + const latest = await publish(); + preparation.reconcileHostPreparation(id); + expect(operations(id)[0]).toMatchObject({ kind: 'library.download', payload: { libraryVersionId: latest.id } }); + verified(id, latest); + hosts.reportHostOperation(id, operations(id)[0].id, { state: 'completed' }); + heartbeat(id, ready); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(2); + vi.setSystemTime(failedAt + 60_000); + heartbeat(id, ready); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(3); + expect(operations(id)[0]).toMatchObject({ kind: 'plutonium.prepare', payload: { revision: 5355 } }); + }); + + it('waits for required library capacity before consuming space for the optional cache', async () => { + const version = await publish(); + const id = enroll({ ...capable, runtimeImages: [runtime], storageAvailable: 1_000 + version.sizeBytes - 1 }); + hosts.updateHost(id, { minimumFreeBytes: 1_000 }); + expect(preparation.reconcileHostPreparation(id).state).toBe('blocked'); + expect(operations(id)).toEqual([]); + }); + + it.each([null, [], {}, { revision: 0, bytes: 1 }, { revision: 1.5, bytes: 1 }, { revision: 1, bytes: -1 }, { revision: 1, bytes: 0.5 }])( + 'rejects invalid cache inventory %j', cache => { + const id = enroll(); + expect(() => heartbeat(id, { plutoniumCache: cache as HostInventory['plutoniumCache'] })).toThrow('Invalid Plutonium cache inventory'); + } + ); + + it('retains only revision and byte count from cache inventory and clears an omitted cache', () => { + const id = enroll({ plutoniumCache: { revision: 5354, bytes: 0, localPath: '/private/cache' } as HostInventory['plutoniumCache'] }); + expect(hosts.getHost(id)?.inventory.plutoniumCache).toEqual({ revision: 5354, bytes: 0 }); + heartbeat(id); + expect(hosts.getHost(id)?.inventory.plutoniumCache).toBeUndefined(); + }); +});