diff --git a/README.md b/README.md index af76cc0..1431af8 100644 --- a/README.md +++ b/README.md @@ -33,13 +33,14 @@ including SQLite metadata and published game-file storage. folder, including its DLC files. Name the library and version, then publish it. Uploads work before any host is enrolled. 2. **Hosts → Add Host**: configure a Debian 12/13 amd64 machine and run its - generated enrollment command. Automatic agent updates are enabled by default. -3. Prepare a Plutainer runtime from the host page and download the published - game-file version to that host from **Game Files**. -4. Add the exact Plutonium server keys in **Keys**, or supply missing keys when + generated enrollment command. The host automatically prepares its Plutainer + runtime and the latest published version of every game library. The host page + shows setup progress. Automatic agent updates are enabled by default. +3. Add the exact Plutonium server keys in **Keys**, or supply missing keys when launching. -5. Launch a server or collection with an explicit host, game-file version and - runtime image. EC reserves host-specific ports and verifies startup. +4. Launch a server or collection with a host and game-file version. The default + runtime is selected automatically. EC reserves host-specific ports and verifies + startup. See [Native host management](docs/native-hosts.md) for installation, credentials, storage, failure recovery, API contracts and the live-host acceptance checklist. @@ -82,8 +83,13 @@ Game-file versions contain the uploaded T5/T6 folder, with its directory layout and included DLC preserved. There is no separate DLC declaration. Versions record the game, label, file manifest, checksums and size. Uploads use bounded chunks, support pause/resume and verify checksums before publication. Actual game files -use dedicated filesystem storage, never SQLite blobs. Host downloads resume, -report progress and verify checksums before becoming usable. +use dedicated filesystem storage, never SQLite blobs. Publishing a new version +automatically prepares it on active hosts, including hosts enrolled later. Host +downloads transfer up to four files concurrently, resume interrupted transfers, +report progress and verify checksums before becoming usable. Failed preparation +retries automatically with backoff; the host page reports any blocking condition. +Set `PLUTAINER_IMAGE` to change the default runtime prepared on hosts. Existing +servers retain their selected library version and runtime. Multiple instances on one host share read-only base files. Scripts, plugins, configuration baselines and overlays retain their versioning, diff --git a/agent/ec_agent.py b/agent/ec_agent.py index ff5517e..0ae344f 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -7,6 +7,7 @@ unknown and is reported for reconciliation, never automatically executed twice. import argparse import base64 +from concurrent.futures import ThreadPoolExecutor, as_completed import contextlib import hashlib import fcntl @@ -29,9 +30,10 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.2.0" +VERSION = "1.2.1" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 +LIBRARY_DOWNLOAD_WORKERS = 4 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") @@ -788,13 +790,46 @@ class Agent: pass libraries, caches = [], [] for marker in (self.root / "libraries").glob("*/*/manifest.json"): + # A crash between writing a manifest and its atomic promotion must + # never expose the staging directory as a ready library. + if marker.parent.name.startswith("."): + continue try: value = json.loads(marker.read_text()) libraries.append({"id": value["libraryId"], "state": "ready"}) caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}) except (OSError, ValueError, KeyError): continue - progress = list(self.library_progress.values()) + ready = {item["id"] for item in libraries} + progress = {key: value.copy() for key, value in list(self.library_progress.items()) if key not in ready} + for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): + try: + value = json.loads(marker.read_text()) + library_id = identifier(value["libraryId"]) + if library_id in ready or library_id in progress: + continue + expected = self.library_dir(value["game"], library_id).with_name("." + library_id + ".partial") + if marker.parent != expected or marker.is_symlink() or marker.parent.is_symlink(): + continue + files = value["files"] + canonical = json.dumps([{ "path": item["path"], "size": item["size"], "sha256": item["sha256"] } for item in files], separators=(",", ":"), ensure_ascii=False) + if digest(canonical.encode("utf-8")) != value["manifestSha256"]: + continue + retained, size, paths = 0, 0, set() + for item in files: + if item["path"] in paths or type(item["size"]) is not int or item["size"] < 0: + raise AgentError("Invalid partial library manifest") + paths.add(item["path"]) + target = scoped(marker.parent / "content", item["path"]) + size += item["size"] + if target.exists() and stat_module.S_ISREG(target.stat().st_mode): + retained += min(target.stat().st_size, item["size"]) + # These are resumable bytes, not verified content. Reporting + # them survives restarts without requiring space for two copies. + progress[library_id] = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesTotal": size, "manifestSha256": value["manifestSha256"]} + except (AgentError, OSError, ValueError, KeyError, TypeError): + continue + progress = list(progress.values()) libraries.extend({"id": item["libraryVersionId"], "state": item["status"]} for item in progress) caches.extend(progress) images = [] @@ -1195,6 +1230,65 @@ class Agent: raise AgentError("Artifact redirects are not allowed") return urllib.request.build_opener(NoRedirect()).open(req, timeout=60) + def download_library_file(self, item, target, files_url, update_progress, aborted): + def check_active(): + if aborted.is_set(): + raise AgentError("Library download stopped after a transfer failed") + + check_active() + target.parent.mkdir(parents=True, exist_ok=True) + offset = target.stat().st_size if target.exists() else 0 + if offset > item["size"]: + target.unlink() + offset = 0 + checksum = hashlib.sha256() + # Hash the saved prefix once so resumed transfers still verify the whole + # file. New bytes are hashed as they arrive, avoiding a second disk read. + if offset: + with target.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + check_active() + checksum.update(chunk) + if offset == item["size"] and target.exists(): + if checksum.hexdigest() == item["sha256"]: + update_progress(item["path"], offset) + return + target.unlink() + offset = 0 + checksum = hashlib.sha256() + update_progress(item["path"], offset) + check_active() + url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") + with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response: + if response.status == 206: + expected = f"bytes {offset}-{item['size'] - 1}/{item['size']}" + if response.headers.get("Content-Range") != expected: + raise AgentError("Invalid resumed download response") + elif response.status != 200: + raise AgentError("Invalid library download response") + elif offset: + # A server may ignore Range and send the whole file instead. + offset = 0 + checksum = hashlib.sha256() + update_progress(item["path"], 0) + with target.open("ab" if offset else "wb") as output: + received = offset + while True: + check_active() + chunk = response.read(1024 * 1024) + if not chunk: + break + received += len(chunk) + if received > item["size"]: + raise AgentError("Library download exceeded its manifest size") + output.write(chunk) + checksum.update(chunk) + update_progress(item["path"], received) + output.flush() + os.fsync(output.fileno()) + if target.stat().st_size != item["size"] or checksum.hexdigest() != item["sha256"]: + raise AgentError("Downloaded library file failed integrity verification") + def library_download(self, payload): library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId"))) with self.artifact_open("GET", payload["manifestUrl"]) as response: @@ -1221,55 +1315,46 @@ class Agent: source_marker = staging / "source.json" if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha: raise Rejected("Partial library download belongs to a different manifest") - atomic_json(source_marker, {"libraryId": library_id, "manifestSha256": manifest_sha}) size = sum(item["size"] for item in files) - downloaded = 0 - paths = set() + atomic_json(source_marker, {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha}) + targets = {} progress = {"libraryVersionId": library_id, "status": "downloading", "bytesDownloaded": 0, "bytesTotal": size, "manifestSha256": manifest_sha} self.library_progress[library_id] = progress try: if shutil.disk_usage(staging).free < max(0, size - sum(path.stat().st_size for path in staging.rglob("*") if path.is_file())): raise AgentError("Insufficient library storage") + # Validate every path before starting concurrent writes. In addition + # to duplicates, a manifest cannot treat a file as a parent folder. for item in files: - if item["path"] in paths or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): + if item["path"] in targets or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): raise AgentError("Invalid library file manifest") - paths.add(item["path"]) - target = scoped(staging / "content", item["path"]) - target.parent.mkdir(parents=True, exist_ok=True) - offset = target.stat().st_size if target.exists() else 0 - if offset > item["size"]: - target.unlink() - offset = 0 - if offset == item["size"] and target.exists() and file_digest(target) != item["sha256"]: - target.unlink() - offset = 0 - if offset < item["size"] or not target.exists(): - url = payload["filesBaseUrl"].rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") - with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response: - if offset and response.status == 206: - content_range = response.headers.get("Content-Range", "") - if not content_range.startswith(f"bytes {offset}-"): - raise AgentError("Invalid resumed download response") - elif offset: - offset = 0 - with target.open("ab" if offset else "wb") as output: - received = offset - while True: - chunk = response.read(1024 * 1024) - if not chunk: - break - received += len(chunk) - if received > item["size"]: - raise AgentError("Library download exceeded its manifest size") - output.write(chunk) - progress["bytesDownloaded"] = downloaded + received - output.flush() - os.fsync(output.fileno()) - progress["status"] = "verifying" - if target.stat().st_size != item["size"] or file_digest(target) != item["sha256"]: - raise AgentError("Downloaded library file failed integrity verification") - downloaded += item["size"] - progress.update(status="downloading", bytesDownloaded=downloaded) + targets[item["path"]] = scoped(staging / "content", item["path"]) + for path in targets: + if any(str(parent) in targets for parent in PurePosixPath(path).parents): + raise AgentError("Library file paths overlap") + progress_lock = threading.Lock() + offsets = {} + aborted = threading.Event() + + def update_progress(path, offset): + with progress_lock: + progress["bytesDownloaded"] += offset - offsets.get(path, 0) + offsets[path] = offset + + def transfer(item): + try: + self.download_library_file(item, targets[item["path"]], payload["filesBaseUrl"], update_progress, aborted) + except Exception: + aborted.set() + raise + + # Bound sockets, memory and disk pressure while allowing other files + # to progress during network round trips and per-file fsyncs. + with ThreadPoolExecutor(max_workers=min(LIBRARY_DOWNLOAD_WORKERS, len(files))) as pool: + futures = [pool.submit(transfer, item) for item in files] + for future in as_completed(futures): + future.result() + progress["status"] = "verifying" saved = {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha} atomic_json(staging / "manifest.json", saved) for path in (staging / "content").rglob("*"): @@ -1446,6 +1531,16 @@ 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": + # 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. + self.journal.update(operation["id"], "running") + try: + result = self.dispatch(operation) + return self.journal.update(operation["id"], "completed", {**result, "reconciled": True}) + except Exception: + return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed") if kind == "library.import": try: with self.artifact_open("GET", payload["completeUrl"]) as response: diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 2739da8..a0d03ac 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -4,6 +4,7 @@ import io import json from pathlib import Path import tempfile +import threading import unittest from unittest.mock import patch @@ -107,6 +108,40 @@ class AgentTests(unittest.TestCase): self.assertEqual(result["state"], "unknown") self.assertEqual(self.docker.calls, []) + def test_agent_restart_resumes_runtime_pull_even_when_an_older_tag_is_cached(self): + image = "ghcr.io/ayymoss/plutainer:latest" + operation = self.operation("runtime.pull", {"image": image}) + self.agent.journal.accept(operation) + self.agent.journal.update(operation["id"], "running") + aliases = self.agent.root / "agent/runtime-images.json" + module.atomic_json(aliases, {image: "sha256:older-image"}) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + original = self.docker.call + + def call(*args, **kwargs): + result = original(*args, **kwargs) + return "sha256:new-image" if args[:2] == ("image", "inspect") else result + + self.docker.call = call + result = self.agent.execute(operation) + self.assertEqual(result["state"], "completed") + self.assertEqual(json.loads(result["result"]), {"image": image, "ready": True, "reconciled": True}) + self.assertEqual(json.loads(aliases.read_text())[image], "sha256:new-image") + self.agent.execute(operation) + self.assertEqual(sum(args[:2] == ("image", "pull") for args in self.docker.calls), 1) + + def test_interrupted_runtime_pull_failure_is_reported_for_automatic_retry(self): + operation = self.operation("runtime.pull", {"image": "ghcr.io/ayymoss/plutainer:latest"}) + self.agent.journal.accept(operation) + self.agent.journal.update(operation["id"], "running") + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + with patch.object(self.docker, "call", side_effect=AgentError("Registry unavailable")): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "failed") + self.assertIn("runtime download", result["error"]) + def test_occupied_and_unknown_wait_then_revalidate_same_operation(self): self.create() operation = self.operation("instance.restart") @@ -198,8 +233,192 @@ class AgentTests(unittest.TestCase): module.scoped(app, path) def manifest(self, content=b"complete-game-file"): - files = [{"path": "zone/test.ff", "size": len(content), "sha256": module.digest(content)}] - return {"id": "downloaded", "game": "t6", "files": files, "sizeBytes": len(content), "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())} + return self.manifest_files({"zone/test.ff": content}) + + def manifest_files(self, contents): + files = [{"path": path, "size": len(content), "sha256": module.digest(content)} for path, content in contents.items()] + return {"id": "downloaded", "game": "t6", "files": files, "sizeBytes": sum(item["size"] for item in files), "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())} + + def test_downloads_use_bounded_parallel_transfers_and_aggregate_progress(self): + contents = {f"zone/file-{index}.ff": bytes([index]) * 31 for index in range(8)} + manifest = self.manifest_files(contents) + barrier = threading.Barrier(4) + lock = threading.Lock() + active = maximum = 0 + snapshots = [] + agent = self.agent + + class ConcurrentResponse(Response): + def __exit__(self, *args): + nonlocal active + with lock: + active -= 1 + snapshots.append(agent.library_progress["downloaded"]["bytesDownloaded"]) + return super().__exit__(*args) + + def fetch(method, path, data=None, extra_headers=None): + nonlocal active, maximum + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + with lock: + active += 1 + maximum = max(maximum, active) + # A serial downloader cannot reach this barrier. Two batches also + # prove that an unlimited worker pool cannot open all eight files. + barrier.wait(timeout=5) + return ConcurrentResponse(contents[path.removeprefix("/files/")]) + + self.agent.artifact_open = fetch + result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertEqual(maximum, 4) + self.assertEqual(snapshots, sorted(snapshots)) + self.assertEqual(max(snapshots), manifest["sizeBytes"]) + self.assertEqual(result["size"], manifest["sizeBytes"]) + for path, content in contents.items(): + self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content" / path).read_bytes(), content) + + def test_fresh_download_hashes_stream_without_rereading_whole_file(self): + content = b"game-file" * 200000 + manifest = self.manifest(content) + self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content) + with patch.object(module, "file_digest", side_effect=AssertionError("download reread the file")): + result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertEqual(result["state"], "ready") + + def test_ignored_range_restarts_file_and_checksum(self): + content = b"complete-game-file" + manifest = self.manifest(content) + staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" + staging.mkdir(parents=True) + (staging / "test.ff").write_bytes(b"bad-prefix") + + def fetch(method, path, data=None, extra_headers=None): + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + self.assertEqual(extra_headers, {"Range": "bytes=10-"}) + return Response(content) + + self.agent.artifact_open = fetch + result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertEqual(result["state"], "ready") + self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) + + def test_resumed_download_verifies_saved_prefix(self): + content = b"complete-game-file" + manifest = self.manifest(content) + staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" + staging.mkdir(parents=True) + (staging / "test.ff").write_bytes(b"corrupt!") + self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode()) if path == "/manifest" else Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) + with self.assertRaisesRegex(AgentError, "integrity"): + self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) + + def test_parallel_failure_retains_partial_files_for_verified_resume(self): + contents = {"zone/first.ff": b"first-game-file", "zone/second.ff": b"second-game-file"} + manifest = self.manifest_files(contents) + payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} + + def failed_fetch(method, path, data=None, extra_headers=None): + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + return Response(contents[path.removeprefix("/files/")][:5]) + + self.agent.artifact_open = failed_fetch + with self.assertRaises(AgentError): + self.agent.library_download(payload) + self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) + self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed") + resumed = [] + + def resume_fetch(method, path, data=None, extra_headers=None): + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + content = contents[path.removeprefix("/files/")] + if extra_headers: + self.assertEqual(extra_headers, {"Range": "bytes=5-"}) + resumed.append(path) + return Response(content[5:], 206, {"Content-Range": f"bytes 5-{len(content)-1}/{len(content)}"}) + return Response(content) + + self.agent.artifact_open = resume_fetch + self.assertEqual(self.agent.library_download(payload)["state"], "ready") + self.assertTrue(resumed) + self.assertEqual(self.agent.verify_library(self.agent.library_dir("t6", "downloaded"))["size"], manifest["sizeBytes"]) + + def test_overlapping_paths_rejected_before_concurrent_downloads(self): + manifest = self.manifest_files({"zone": b"file", "zone/child.ff": b"child"}) + calls = [] + + def fetch(method, path, data=None, extra_headers=None): + calls.append(path) + return Response(json.dumps(manifest).encode()) + + self.agent.artifact_open = fetch + with self.assertRaisesRegex(AgentError, "overlap"): + self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertEqual(calls, ["/manifest"]) + + def test_failed_transfer_restart_reports_retained_bytes_and_resumes_with_only_remaining_space(self): + content = b"complete-game-file" + manifest = self.manifest(content) + payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} + self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content[:8]) + with self.assertRaises(AgentError): + self.agent.library_download(payload) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") + self.assertEqual(cache, {"libraryVersionId": "downloaded", "status": "missing", "bytesDownloaded": 8, "bytesTotal": len(content), "manifestSha256": manifest["manifestSha256"]}) + + def fetch(method, path, data=None, extra_headers=None): + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + self.assertEqual(extra_headers, {"Range": "bytes=8-"}) + return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) + + self.agent.artifact_open = fetch + usage = module.shutil.disk_usage(self.agent.root) + with patch.object(module.shutil, "disk_usage", return_value=type(usage)(usage.total, usage.used, len(content) - 8)): + self.assertEqual(self.agent.library_download(payload)["state"], "ready") + + def test_unpromoted_manifest_is_never_reported_as_ready_and_progress_is_not_duplicated(self): + content = b"complete-game-file" + manifest = self.manifest(content) + staging = self.agent.root / "libraries/t6/.downloaded.partial" + (staging / "content/zone").mkdir(parents=True) + # Oversized retained files can be replaced during resume. Their progress + # must still stay within the published file size. + (staging / "content/zone/test.ff").write_bytes(content + b"extra") + saved = {"libraryId": "downloaded", "game": "t6", "size": len(content), "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]} + module.atomic_json(staging / "source.json", saved) + module.atomic_json(staging / "manifest.json", saved) + inventory = self.agent.inventory() + caches = [item for item in inventory["libraryCaches"] if item["libraryVersionId"] == "downloaded"] + self.assertEqual(len(caches), 1) + self.assertEqual(caches[0]["status"], "missing") + self.assertEqual(caches[0]["bytesDownloaded"], len(content)) + self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"]) + self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3} + caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"] + self.assertEqual(len(caches), 1) + self.assertEqual(caches[0]["status"], "downloading") + self.assertEqual(caches[0]["bytesDownloaded"], 3) + + def test_partial_inventory_rejects_changed_manifest_and_symlinked_content(self): + content = b"complete-game-file" + manifest = self.manifest(content) + staging = self.agent.root / "libraries/t6/.downloaded.partial" + (staging / "content/zone").mkdir(parents=True) + source = {"libraryId": "downloaded", "game": "t6", "files": manifest["files"], "manifestSha256": "0" * 64} + module.atomic_json(staging / "source.json", source) + (staging / "content/zone/test.ff").write_bytes(content) + self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"])) + source["manifestSha256"] = manifest["manifestSha256"] + module.atomic_json(staging / "source.json", source) + (staging / "content/zone/test.ff").unlink() + (staging / "content/zone/test.ff").symlink_to(self.library / "content/game.ff") + self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"])) def test_resume_verifies_and_atomically_publishes_library(self): content = b"complete-game-file" diff --git a/docs/native-hosts.md b/docs/native-hosts.md index 8ced1e2..f0c4813 100644 --- a/docs/native-hosts.md +++ b/docs/native-hosts.md @@ -18,23 +18,39 @@ the agent owns local execution and its durable operation journal. installation, verifies the downloaded agent and updater SHA-256 over HTTPS, enrolls it and enables `electric-cherry-agent.service` and its automatic update timer. It needs no Git, Node.js or local build. -4. In **Hosts**, prepare a Plutainer runtime image. In **Game Files**, select T5 or - T6 and upload your local game folder. Name the library and version and publish +4. In **Game Files**, select T5 or T6 and upload your local game folder. Name the + library and version and publish it after the upload finishes. Include the DLC files in that folder; there is no separate DLC declaration. Uploading also works before enrolling any host. -5. Cache the published version on each destination host. Wait for verification. - Hosts report connection and readiness separately; deployment also checks the - particular library, image, capacity and ports you selected. A connected host - missing content shows **Needs runtime** or **Needs game files**; these require - setup rather than waiting for an automatic preparation step. **Ready** means - it has both a runtime image and a verified library available. +5. Hosts automatically prepare the configured Plutainer runtime and the latest + published version of every game library. **Automatic host setup** reports + progress and errors. Wait for verification before launching. Hosts report + connection and readiness separately; deployment also checks the particular + library, image, capacity and ports you selected. 6. Launch a server or collection. Select the host, immutable game-file version(s) - and runtime image. EC selects lobby IDs and reserves UDP/TCP ports atomically. + and use the automatically selected default runtime, or an advanced override. + EC selects lobby IDs and reserves UDP/TCP ports atomically. Saved Plutonium server keys remain associated with their exact server names. A fresh destination host downloads centrally stored game files without manual copying or per-container setup. EC does not supply the game content itself. +The controller reconciles automatic setup on agent heartbeats and operation +polls. It first pulls `PLUTAINER_IMAGE` (default +`ghcr.io/ayymoss/plutainer:latest`), then prepares the newest published version of +each library, one setup operation at a time. Publishing more 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 operation history. Disconnected hosts resume when they +reconnect. Draining and disabled hosts do not receive new automatic setup work. + +Verified older versions remain cached and running servers keep their selected +files and runtime. Automatic setup ensures that the configured image is present; +it does not refresh an already cached mutable image tag. Advanced controls on the +host and game-file pages can explicitly pull a custom runtime, prepare an older +library version or retry preparation immediately. + The default host root is `/var/lib/electric-cherry`; installer `--data-dir` changes it. Use a dedicated, root-owned directory. For a separate disk, use a child of its mount point (for example `/mnt/games/electric-cherry`), with no nested mounts, so @@ -247,8 +263,12 @@ the page open while transferring. Once publication starts, the controller hashes every complete file and publishes the directory atomically; closing the browser does not stop that verification. An interrupted controller can retry publication. Published versions cannot be updated; -changing files requires a new version. Hosts download with HTTP Range resumption, -report progress, verify all checksums and atomically promote the complete cache. +changing files requires a new version. Hosts download up to four files +concurrently with HTTP Range resumption, report progress, verify all checksums +and atomically promote the complete cache. Checksums are calculated while bytes +arrive, avoiding another complete read of newly downloaded files. Resumed files +include their saved prefix in verification. Progress reports are rate-limited so +controller round trips do not stall every transferred chunk. Corrupt or incomplete content never becomes Ready. Cache inventory is reconciled on heartbeats, including vanished caches. Older immutable versions remain usable when their exact manifest is verified. diff --git a/src/lib/hosts.ts b/src/lib/hosts.ts index 1e429fe..8da339e 100644 --- a/src/lib/hosts.ts +++ b/src/lib/hosts.ts @@ -74,6 +74,7 @@ export interface Host { readiness: HostReadiness; readinessReasons: string[]; inventory: HostInventory; enrolled: boolean; credentialActive: boolean; decommission: HostDecommission | null; + preparation?: HostPreparation; } export interface HostOperation { id: string; hostId: string; kind: string; instanceId: string | null; generation: number | null; @@ -81,6 +82,15 @@ export interface HostOperation { createdAt: string; updatedAt: string; } +export interface HostPreparationItem { + kind: 'runtime' | 'library'; label: string; + state: 'waiting' | 'preparing' | 'ready' | 'blocked'; detail: string; + operationId?: string; bytesDownloaded?: number; bytesTotal?: number; +} +export interface HostPreparation { + state: HostPreparationItem['state']; items: HostPreparationItem[]; +} + export interface NativeDeploymentSelection { hostId: string; libraryIds: Partial>; image: string; bindAddress?: string; publicAddress?: string; publicPort?: number; diff --git a/src/lib/server/game-files/index.ts b/src/lib/server/game-files/index.ts index fca90cf..a8666b0 100644 --- a/src/lib/server/game-files/index.ts +++ b/src/lib/server/game-files/index.ts @@ -38,6 +38,14 @@ export function newestReadyLibrary(game: string): GameLibraryVersion | null { const row = database().prepare("SELECT v.*, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id WHERE l.game=? AND v.state='ready' ORDER BY v.created_at DESC, v.rowid DESC LIMIT 1").get(game) as VersionRow | undefined; return row ? versionFromRow(row) : null; } +/** Preparation needs only the latest published version of each library, never its full manifest. */ +export function latestPublishedLibraries() { + return database().prepare(`SELECT v.id, l.name, l.game, v.version, v.size_bytes AS sizeBytes, v.manifest_sha256 AS manifestSha256 + FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id + WHERE v.state='ready' AND v.id=(SELECT newest.id FROM game_file_versions newest + WHERE newest.library_id=l.id AND newest.state='ready' ORDER BY newest.created_at DESC,newest.rowid DESC LIMIT 1) + ORDER BY l.name`).all() as { id: string; name: string; game: GameLibraryGame; version: string; sizeBytes: number; manifestSha256: string }[]; +} export function listLibraries(): GameLibrary[] { reconcileLibraryOperations(); return (database().prepare('SELECT * FROM game_file_libraries ORDER BY name').all() as Array<{ id: string; name: string; game: GameLibraryGame; description: string; created_at: string }>).map(row => ({ @@ -315,18 +323,19 @@ export function queueLibraryDownload(hostId: string, versionId: string) { if (!version || version.state !== 'ready') throw new Error('Choose a published library version.'); const host = getHost(hostId); if (!host || !host.connected || !host.credentialActive || host.mode === 'disabled') throw new Error('Choose an online, enrolled host.'); - if (Number(host.inventory.storageAvailable ?? 0) < version.sizeBytes + host.minimumFreeBytes) throw new Error('This host has insufficient storage for this library and its minimum free-space reserve.'); const prior = getHostLibraryCache(hostId, versionId); if (prior?.operationId) { const operation = getHostOperation(prior.operationId); if (operation && ['pending', 'accepted', 'running', 'unknown'].includes(operation.state)) return operation; } + const retainedBytes = prior?.manifestSha256 === version.manifestSha256 ? prior.bytesDownloaded : 0; + if (Number(host.inventory.storageAvailable ?? 0) < version.sizeBytes - retainedBytes + host.minimumFreeBytes) throw new Error('This host has insufficient storage for this library and its minimum free-space reserve.'); const operation = queueHostOperation(hostId, 'library.download', { libraryVersionId: versionId, game: version.game, manifestUrl: `/api/agent/libraries/${versionId}/manifest`, filesBaseUrl: `/api/agent/libraries/${versionId}/files`, statusUrl: `/api/agent/libraries/${versionId}/status`, manifestSha256: version.manifestSha256, sizeBytes: version.sizeBytes }); - updateHostLibraryCaches(hostId, [{ libraryVersionId: versionId, status: 'downloading', bytesDownloaded: 0 }]); + updateHostLibraryCaches(hostId, [{ libraryVersionId: versionId, status: 'downloading', bytesDownloaded: retainedBytes, manifestSha256: prior?.manifestSha256 }]); database().prepare('UPDATE host_library_caches SET operation_id=? WHERE host_id=? AND version_id=?').run(operation.id, hostId, versionId); return operation; } diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index c9874a9..dabf90a 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -33,8 +33,8 @@ function hostFromRow(row: HostRow): Host { const infrastructureReady = reasons.length === 0; const runtimeReady = !!inventory.runtimeImages?.length; const libraryReady = !!inventory.libraries?.some(library => library.state === 'ready'); - if (!runtimeReady) reasons.push('Prepare a Plutainer runtime image.'); - if (!libraryReady) reasons.push('Download and verify a game-file library.'); + if (!runtimeReady) reasons.push('The Plutainer runtime image is not ready yet.'); + if (!libraryReady) reasons.push('No verified game-file library is available yet.'); if (row.mode === 'draining') reasons.push('Host is draining; new deployments are blocked.'); if (row.mode === 'disabled') reasons.push('Host is disabled.'); const decommission = presentDecommission(decommissionRow(row.id)); diff --git a/src/lib/server/hosts/preparation.ts b/src/lib/server/hosts/preparation.ts new file mode 100644 index 0000000..64bada1 --- /dev/null +++ b/src/lib/server/hosts/preparation.ts @@ -0,0 +1,123 @@ +import { db } from '../db/index.js'; +import { config } from '../config.js'; +import { AGENT_PROTOCOL_VERSION, getHost, queueHostOperation } from './index.js'; +import { getHostLibraryCache, latestPublishedLibraries, queueLibraryDownload } from '../game-files/index.js'; +import type { Host, HostOperationState, HostPreparation, HostPreparationItem } from '../../hosts.js'; + +const activeStates = ['pending', 'accepted', 'running', 'unknown']; +type PreparationOperation = { id: string; state: HostOperationState; updated_at: string; error: string | null }; +type Target = { item: HostPreparationItem; image?: string; versionId?: string; requiredBytes: number; operation?: PreparationOperation }; + +function latestOperation(hostId: string, kind: string, field: string, value: string): PreparationOperation | undefined { + return db().prepare(`SELECT id,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; +} + +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!; + // 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' + 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 }; + return Date.parse(operation.updated_at) + Math.min(15 * 60_000, 60_000 * 2 ** Math.min(4, Math.max(0, count - 1))); +} + +function hostBlocker(host: Host): { state: 'waiting' | 'blocked'; detail: string } | null { + if (host.decommission) return { state: 'waiting', detail: 'Automatic preparation is paused during host deletion.' }; + if (host.mode !== 'active') return { state: 'waiting', detail: `Automatic preparation is paused while this host is ${host.mode}.` }; + if (!host.enrolled) return { state: 'waiting', detail: 'Automatic preparation starts after agent installation.' }; + if (!host.connected || !host.credentialActive) return { state: 'waiting', detail: 'Waiting for the host to reconnect.' }; + if (host.inventory.protocolVersion !== AGENT_PROTOCOL_VERSION) return { state: 'blocked', detail: 'A compatible host agent is required.' }; + if (!host.inventory.dockerRunning) return { state: 'blocked', detail: 'Waiting for Docker to become available on the host.' }; + if (!host.inventory.directoriesWritable) return { state: 'blocked', detail: 'The host cannot write its managed directories.' }; + return null; +} + +function targets(host: Host): Target[] { + const image = config.komodo.defaultImage; + const runtimeReady = host.inventory.runtimeImages?.includes(image); + const result: Target[] = [{ image, requiredBytes: host.minimumFreeBytes, + operation: latestOperation(host.id, 'runtime.pull', 'image', image), + item: { kind: 'runtime', label: image, state: runtimeReady ? 'ready' : 'waiting', detail: runtimeReady ? 'Runtime is cached.' : 'Runtime will be downloaded automatically.' } }]; + for (const version of latestPublishedLibraries()) { + const cache = getHostLibraryCache(host.id, version.id); + const matches = cache?.manifestSha256 === version.manifestSha256; + const ready = matches && ['ready', 'outdated'].includes(cache!.status) && cache!.bytesDownloaded === version.sizeBytes; + result.push({ versionId: version.id, requiredBytes: version.sizeBytes - (matches ? cache!.bytesDownloaded : 0) + host.minimumFreeBytes, + operation: latestOperation(host.id, 'library.download', 'libraryVersionId', version.id), + item: { kind: 'library', label: `${version.name} · ${version.version}`, state: ready ? 'ready' : 'waiting', + detail: ready ? 'Game files are downloaded and verified.' : 'Game files will be downloaded and verified automatically.', + bytesDownloaded: cache?.bytesDownloaded ?? 0, bytesTotal: version.sizeBytes } }); + } + const blocker = hostBlocker(host); + for (const target of result) { + if (target.item.state === 'ready') continue; + const operation = target.operation; + if (operation) target.item.operationId = operation.id; + if (blocker) { Object.assign(target.item, blocker); continue; } + if (operation && activeStates.includes(operation.state)) { + target.item.state = 'preparing'; + target.item.detail = operation.state === 'unknown' ? 'Waiting for the agent to reconcile interrupted preparation.' + : target.image ? 'Downloading the runtime automatically.' : 'Downloading and verifying game files automatically.'; + } 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 (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) { + target.item.state = 'blocked'; target.item.detail = 'Insufficient free storage, including the host reserve. Preparation resumes automatically when space is available.'; + } + } + return result; +} + +export function hostPreparationStatus(host: Host): HostPreparation { + const items = targets(host).map(target => target.item); + if (!items.some(item => item.kind === 'library')) items.push({ kind: 'library', label: 'Game files', state: 'waiting', + detail: 'Upload and publish a game-file library once; hosts download it 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'; + 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, preparation }; +} + +/** Agent heartbeats and polls drive durable preparation; no open UI or in-memory worker is needed. */ +export function reconcileHostPreparation(hostId: string): HostPreparation { + return db().transaction(() => { + const host = getHost(hostId); + if (!host) throw new Error('Host was not found.'); + if (!hostBlocker(host)) { + // Queue only one setup mutation at a time so each capacity check uses + // fresh disk inventory. This also respects manually queued downloads. + 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') + ORDER BY updated_at DESC LIMIT 1`).get(hostId) 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'); + if (target?.image) queueHostOperation(host.id, 'runtime.pull', { image: target.image }); + else if (target?.versionId) queueLibraryDownload(host.id, target.versionId); + } + } + return hostPreparationStatus(host); + }).immediate(); +} diff --git a/src/lib/ui/components/DeployCollectionDialog.svelte b/src/lib/ui/components/DeployCollectionDialog.svelte index c2cc938..258871b 100644 --- a/src/lib/ui/components/DeployCollectionDialog.svelte +++ b/src/lib/ui/components/DeployCollectionDialog.svelte @@ -17,7 +17,7 @@ let mode = $state('fill'); let games = $state([]); let collectionName = $state('Launch collection'); - let hostSelection = $state({ hostId: '', libraryIds: {}, image: 'ghcr.io/ayymoss/plutainer:latest' }); + let hostSelection = $state({ hostId: '', libraryIds: {}, image: '' }); let hostReady = $state(false); let confirmRemovals = $state(false); let keys = $state>({}); @@ -41,7 +41,7 @@ async function initialize() { const current = ++session; busy = true; error = ''; games = []; hostReady = false; - hostSelection = { hostId: '', libraryIds: {}, image: 'ghcr.io/ayymoss/plutainer:latest' }; + hostSelection = { hostId: '', libraryIds: {}, image: '' }; try { const response = await fetch('/api/blueprints'); const body = await response.json(); if (!response.ok) throw new Error(body.error ?? 'Could not load collection.'); diff --git a/src/lib/ui/components/DeployServerDialog.svelte b/src/lib/ui/components/DeployServerDialog.svelte index 907a115..0481950 100644 --- a/src/lib/ui/components/DeployServerDialog.svelte +++ b/src/lib/ui/components/DeployServerDialog.svelte @@ -23,7 +23,7 @@ let maps = $derived(catalogs[game]); let selectedBlueprint = $derived(blueprints.find(b => b.id === blueprintId)); let preview = $state(null); - let hostSelection = $state({ hostId: '', libraryIds: {}, image: 'ghcr.io/ayymoss/plutainer:latest' }); + let hostSelection = $state({ hostId: '', libraryIds: {}, image: '' }); let hostReady = $state(false); let plutoKey = $state(''); let busy = $state(false); @@ -54,7 +54,7 @@ blueprintId = initialBlueprintId; game = initialGame; mapSlug = initialMapSlug; - hostSelection = { hostId: '', libraryIds: {}, image: 'ghcr.io/ayymoss/plutainer:latest' }; hostReady = false; + hostSelection = { hostId: '', libraryIds: {}, image: '' }; hostReady = false; try { const [mapRes, blueprintRes] = await Promise.all([fetch('/api/maps'), fetch('/api/blueprints')]); if (!mapRes.ok || !blueprintRes.ok) throw new Error('Could not load maps and collections. Try again.'); diff --git a/src/lib/ui/components/HostPreparation.svelte b/src/lib/ui/components/HostPreparation.svelte new file mode 100644 index 0000000..0ebd9f6 --- /dev/null +++ b/src/lib/ui/components/HostPreparation.svelte @@ -0,0 +1,31 @@ + + +
+

The default runtime and the latest published version of each game library are prepared automatically on active hosts. New versions leave running servers on their selected files.

+ {#if host.preparation} +

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

+
    + {#each host.preparation.items as item} +
  • +
    {item.label}{item.state === 'blocked' ? 'Needs attention' : item.state}
    +

    {item.detail}

    + {#if item.bytesTotal && item.bytesDownloaded !== undefined && item.state === 'preparing'} + + {formatGameBytes(item.bytesDownloaded)} / {formatGameBytes(item.bytesTotal)} + {/if} +
    • + {/each} +
+ {:else}

Setup progress will appear after the host connects.

{/if} +

Upload and view game-file libraries →

+
+ + diff --git a/src/lib/ui/components/NativeHostSelection.svelte b/src/lib/ui/components/NativeHostSelection.svelte index d408e74..45c26fc 100644 --- a/src/lib/ui/components/NativeHostSelection.svelte +++ b/src/lib/ui/components/NativeHostSelection.svelte @@ -4,36 +4,58 @@ import { hostReadinessLabel } from '$lib/hosts.js'; import type { GameLibrary } from '$lib/game-files.js'; import type { ZombieGame } from '$lib/map-aliases.js'; + import { pollVisible } from '$ui/poll.js'; let { games, selection = $bindable(), ready = $bindable(false), disabled = false, network = false, onchange }: { games: ZombieGame[]; selection: NativeDeploymentSelection; ready?: boolean; disabled?: boolean; network?: boolean; onchange?: () => void } = $props(); let hosts = $state([]); let libraries = $state([]); let loading = $state(true); let error = $state(''); + let defaultRuntimeImage = $state(''); + let runtimeSelection: NativeDeploymentSelection | null = null; const host = $derived(hosts.find(item => item.id === selection.hostId)); - const versions = $derived(libraries.flatMap(library => library.versions.filter(version => version.state === 'ready').map(version => ({ ...version, name: library.name })))); + // Keep each library's server ordering, including its tie breaker for versions + // created together. Publication completion time does not choose auto setup. + const versions = $derived(libraries.flatMap(library => library.versions.filter(version => version.state === 'ready').map(version => ({ ...version, name: library.name }))).sort((a, b) => b.createdAt.localeCompare(a.createdAt))); + const latestVersions = $derived(libraries.flatMap(library => library.versions.filter(version => version.state === 'ready').slice(0, 1)).sort((a, b) => b.createdAt.localeCompare(a.createdAt))); const imageReady = $derived(!!host?.inventory.runtimeImages?.includes(selection.image)); function cacheReady(versionId: string | undefined) { return !!versionId && versions.some(version => version.id === versionId && version.caches.some(cache => cache.hostId === selection.hostId && ['ready', 'outdated'].includes(cache.status))); } - $effect(() => { ready = !loading && host?.readiness === 'ready' && imageReady && games.length > 0 && games.every(game => cacheReady(selection.libraryIds[game])); }); + $effect(() => { ready = !loading && !error && host?.readiness === 'ready' && imageReady && games.length > 0 && games.every(game => cacheReady(selection.libraryIds[game])); }); + $effect(() => { + if (disabled || !defaultRuntimeImage || runtimeSelection === selection) return; + if (!selection.image) selection.image = defaultRuntimeImage; + runtimeSelection = selection; + }); + $effect(() => { + if (!host || disabled) return; + let changed = false; + for (const game of games) { + if (selection.libraryIds[game]) continue; + const latest = latestVersions.find(version => version.game === game); + if (latest) { selection.libraryIds[game] = latest.id; changed = true; } + } + if (changed) onchange?.(); + }); async function load() { - loading = true; error = ''; try { const [hostResponse, libraryResponse] = await Promise.all([fetch('/api/hosts'), fetch('/api/game-files')]); if (!hostResponse.ok || !libraryResponse.ok) throw new Error('Could not load hosts and game libraries.'); - const [hostBody, libraryBody] = await Promise.all([hostResponse.json(), libraryResponse.json()]); hosts = hostBody.hosts; libraries = libraryBody.libraries; + const [hostBody, libraryBody] = await Promise.all([hostResponse.json(), libraryResponse.json()]); hosts = hostBody.hosts; libraries = libraryBody.libraries; error = ''; + defaultRuntimeImage = hostBody.defaultRuntimeImage ?? ''; } catch (cause) { error = (cause as Error).message; } finally { loading = false; } } - onMount(() => { void load(); }); + onMount(() => pollVisible(load, 5000));

Host and game files

{#if error}{/if} {#if loading}

Loading host readiness and libraries…

{/if} - - {#if !loading && !hosts.some(candidate => candidate.readiness === 'ready')}

No deployment-ready hosts. Set up a host, then refresh.

{/if} + + {#if !loading && !hosts.length}

Add a host. Its runtime and published game files will be prepared automatically.

{/if} {#if host}

{host.hostname} · ports {host.portMin}–{host.portMax} · {host.publicAddress || host.bindAddress}

{/if} - {#each games as game}{/each} -

Upload a game folder or download a version to a host, then refresh. Each launch stays pinned to its selected library version.

- - {#if host && !imageReady}

Prepare this runtime image from host setup before launch.

{/if} + {#if host?.preparation && host.preparation.state !== 'ready'}

{host.preparation.items.find(item => item.state === 'blocked' || item.state === 'preparing' || item.state === 'waiting')?.detail} View automatic setup

{/if} + {#each games as game}{#if host && selection.libraryIds[game] && !cacheReady(selection.libraryIds[game])}

The selected files must finish preparation before launch. The latest published versions download automatically; older versions can be prepared from Game Files.

{/if}{/each} +

Publish a game folder to make it available automatically on hosts. Each launch stays pinned to its selected version. Setup progress refreshes automatically.

+
Advanced: runtime override

The configured default runtime is prepared automatically. Custom images must be prepared from the host page.

+ {#if host && !imageReady}

Waiting for runtime preparation. View host setup.

{/if} {#if network}
Network overrides

Public mapping does not change the local Docker port reservation.

{/if}
diff --git a/src/routes/api/agent/heartbeat/+server.ts b/src/routes/api/agent/heartbeat/+server.ts index f405084..ffb312a 100644 --- a/src/routes/api/agent/heartbeat/+server.ts +++ b/src/routes/api/agent/heartbeat/+server.ts @@ -1,6 +1,7 @@ import { json } from '@sveltejs/kit'; import { heartbeat } from '$lib/server/hosts/index.js'; import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; +import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; export async function POST({ request }: { request: Request }) { const authenticated = authenticateAgent(request); try { @@ -11,6 +12,7 @@ export async function POST({ request }: { request: Request }) { const { updateHostLibraryCaches } = await import('$lib/server/game-files/index.js'); updateHostLibraryCaches(host.id, inventory.libraryCaches, { snapshot: true }); } + reconcileHostPreparation(host.id); return json({ host, heartbeatSeconds: 15, serverTime: new Date().toISOString() }); } catch (cause) { return hostApiError(cause); } } diff --git a/src/routes/api/agent/operations/+server.ts b/src/routes/api/agent/operations/+server.ts index 1bc3e2e..e6c0b74 100644 --- a/src/routes/api/agent/operations/+server.ts +++ b/src/routes/api/agent/operations/+server.ts @@ -1,7 +1,9 @@ import { json } from '@sveltejs/kit'; import { pendingHostOperations } from '$lib/server/hosts/index.js'; import { authenticateAgent } from '$lib/server/hosts/http.js'; +import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; export function GET({ request }: { request: Request }) { const host = authenticateAgent(request); + reconcileHostPreparation(host.id); return json({ operations: pendingHostOperations(host.id), serverTime: new Date().toISOString() }, { headers: { 'Cache-Control': 'no-store' } }); } diff --git a/src/routes/api/hosts/+server.ts b/src/routes/api/hosts/+server.ts index 4d37409..f42396f 100644 --- a/src/routes/api/hosts/+server.ts +++ b/src/routes/api/hosts/+server.ts @@ -1,8 +1,10 @@ import { json } from '@sveltejs/kit'; import { createEnrollment, createHost, listHosts } from '$lib/server/hosts/index.js'; import { hostApiError, installerCommand, objectBody, requireManagementOrigin } from '$lib/server/hosts/http.js'; +import { presentHostPreparation } from '$lib/server/hosts/preparation.js'; +import { config } from '$lib/server/config.js'; -export function GET({ request }: { request: Request }) { requireManagementOrigin(request); return json({ hosts: listHosts() }); } +export function GET({ request }: { request: Request }) { requireManagementOrigin(request); return json({ hosts: listHosts().map(presentHostPreparation), defaultRuntimeImage: config.komodo.defaultImage }); } export async function POST({ request }: { request: Request }) { requireManagementOrigin(request); try { diff --git a/src/routes/api/hosts/[id]/+server.ts b/src/routes/api/hosts/[id]/+server.ts index a210d6b..25267fb 100644 --- a/src/routes/api/hosts/[id]/+server.ts +++ b/src/routes/api/hosts/[id]/+server.ts @@ -1,12 +1,14 @@ import { error, json } from '@sveltejs/kit'; import { deleteHost, getHost, hostDeletionStatus, HostDeletionError, listHostOperations, updateHost } from '$lib/server/hosts/index.js'; import { hostApiError, objectBody, requireManagementOrigin } from '$lib/server/hosts/http.js'; +import { presentHostPreparation } from '$lib/server/hosts/preparation.js'; +import { config } from '$lib/server/config.js'; type Event = { request: Request; params: { id: string } }; export function GET({ request, params }: Event) { requireManagementOrigin(request); const host = getHost(params.id); if (!host) error(404, 'Host not found.'); // Payloads and results can contain RCON keys or staged credentials. History // exposes metadata only; agent APIs remain the only command delivery path. - return json({ host, deletion: hostDeletionStatus(host.id), operations: listHostOperations(host.id).map(({ payload: _payload, result: _result, ...operation }) => operation) }); + return json({ host: presentHostPreparation(host), defaultRuntimeImage: config.komodo.defaultImage, deletion: hostDeletionStatus(host.id), operations: listHostOperations(host.id).map(({ payload: _payload, result: _result, ...operation }) => operation) }); } export async function PATCH({ request, params }: Event) { requireManagementOrigin(request); diff --git a/src/routes/game-files/+page.svelte b/src/routes/game-files/+page.svelte index 72a648d..3b56cf5 100644 --- a/src/routes/game-files/+page.svelte +++ b/src/routes/game-files/+page.svelte @@ -36,6 +36,14 @@ const progressBytes = $derived(Math.min(totalBytes, acceptedBytes + sendingBytes)); const transferring = $derived(uploadBusy && (phase === 'preparing' || phase === 'uploading')); const hostName = (id: string | null) => id ? hosts.find(host => host.id === id)?.name ?? id : 'Browser upload'; + function automaticDownloadDetail(host: Host | undefined): string { + if (host?.decommission) return 'Automatic setup is paused during host deletion.'; + if (!host?.enrolled) return 'Waiting for host enrollment.'; + if (host.mode !== 'active') return 'Automatic setup resumes when the host is active.'; + if (!host.credentialActive || !host.connected) return 'Automatic setup resumes when the host reconnects.'; + if (host.readiness === 'degraded') return host.readinessReasons[0] ?? 'The host needs attention before automatic setup can continue.'; + return 'Download starts automatically after runtime preparation and earlier work finish.'; + } async function responseBody(response: Response) { let body; @@ -101,7 +109,7 @@ } if (disposed) return; if (status.status !== 'published') throw new Error(status.error || 'Verification did not finish. Retry this upload to continue.'); - phase = 'published'; folder = null; notice = `${name || 'Game files'} ${version} published. You can download this version to a host below.`; + phase = 'published'; folder = null; notice = `${name || 'Game files'} ${version} published. Active hosts will download and verify this version automatically.`; if (folderInput) folderInput.value = ''; await load(); } @@ -192,12 +200,12 @@ Game Files | Electric Cherry
- + {#snippet actions()}{#if canManage}{/if}{/snippet} {#if error}{/if} {#if notice}

{notice}

{/if} - {#if !canManage}

Administrators upload game files and distribute verified copies to hosts.

{/if} + {#if !canManage}

Administrators publish game files. Active hosts prepare verified copies automatically.

{/if} {#if canManage && showUpload}

{uploadId ? phase === 'published' ? 'Version published' : 'Continue folder upload' : 'Upload a game folder'}

@@ -216,7 +224,7 @@ {/if} {#if folder}
{folder.name}{folder.files.length.toLocaleString()} files · {formatGameBytes(folder.size)}
Preview folder paths
    {#each folder.files.slice(0, 8) as file}
  • {file.path}
    • {/each}
{#if folder.files.length > 8}

And {(folder.files.length - 8).toLocaleString()} more files.

{/if}
{/if} {#if phase !== 'idle'}
-
{phase === 'preparing' ? 'Preparing upload…' : phase === 'uploading' ? 'Uploading game files' : phase === 'verifying' ? 'Verifying and publishing…' : phase === 'published' ? 'Published and ready to distribute' : 'Upload paused'}{completedFiles.toLocaleString()} / {totalFiles.toLocaleString()} files
+
{phase === 'preparing' ? 'Preparing upload…' : phase === 'uploading' ? 'Uploading game files' : phase === 'verifying' ? 'Verifying and publishing…' : phase === 'published' ? 'Published · host downloads are automatic' : 'Upload paused'}{completedFiles.toLocaleString()} / {totalFiles.toLocaleString()} files

{formatGameBytes(progressBytes)} / {formatGameBytes(totalBytes)} saved or transferring

{#if currentPath}

{currentPath}

{/if} @@ -232,6 +240,7 @@ {/if}
{libraries.length} libraries · Game folders and runtime images are versioned separately.
{#each matching as library (library.id)} + {@const latestPublished = library.versions.find(item => item.state === 'ready')}

{library.name}

{library.game.toUpperCase()} · {library.description}

{#if canManage}{/if}
{#each library.versions as item (item.id)} @@ -242,11 +251,22 @@ {#if item.error}

{item.error}

{/if}
Version details
Immutable version ID
{item.id}
Manifest SHA256
{item.manifestSha256 ?? 'Available after verification'}
Uploaded from
{browserUpload ? 'Game folder in browser' : `${hostName(item.sourceHostId)} / ${item.sourceVolume}`}
{#if item.state === 'ready'}View complete file manifest{/if}
{#if item.state === 'ready'} - {#if canManage}
{#if !onlineHosts.length}

This version is stored and ready. Connect a host when you want to deploy game servers.

{/if}{/if} + {#if item.id === latestPublished?.id}

This is the latest published version. Active hosts download and verify it automatically, including hosts enrolled later. Running servers keep their selected version.

{:else}

This older version remains available to servers pinned to it. Use the advanced download control to prepare it on another host.

{/if} + {#if !hosts.length}

Files are published. Add a host to prepare them automatically.

{/if}
- {#each item.caches as cache}{:else}{/each} + {#each item.caches as cache} + {@const cacheHost = hosts.find(host => host.id === cache.hostId)} + {@const awaitingAutomaticDownload = item.id === latestPublished?.id && cache.status === 'missing'} + + + + + + + {:else}{/each}
HostCache stateProgressDetail
{hostName(cache.hostId)}{cache.status}{!hosts.find(host => host.id === cache.hostId)?.connected ? ' · offline' : ''}{#if cache.status === 'downloading' || cache.status === 'verifying'}{formatGameBytes(cache.bytesDownloaded)} / {formatGameBytes(cache.bytesTotal)}{:else if cache.status === 'ready' || cache.status === 'outdated'}Verified{cache.status === 'outdated' ? ' · older version' : ''}{:else}—{/if}{cache.error ?? ''}
No host copies yet.
{hostName(cache.hostId)}{awaitingAutomaticDownload ? 'Waiting' : cache.status}{!cacheHost?.connected ? ' · offline' : ''}{#if cache.status === 'downloading' || cache.status === 'verifying'}{formatGameBytes(cache.bytesDownloaded)} / {formatGameBytes(cache.bytesTotal)}{:else if cache.status === 'ready' || cache.status === 'outdated'}Verified{cache.status === 'outdated' ? ' · older version' : ''}{:else}—{/if}{cache.error ?? (awaitingAutomaticDownload ? automaticDownloadDetail(cacheHost) : '')}{#if awaitingAutomaticDownload || cache.status === 'failed'} View host setup{/if}
No host copies yet.
- {:else if browserUpload}

Resume with the original unchanged folder. Published versions can be downloaded to any enrolled host.

{#if canManage}
{/if} + {#if canManage}
Advanced: download or verify this version

Prepare an older version or retry a download now. Latest versions are prepared automatically.

{/if} + {:else if browserUpload}

Resume with the original unchanged folder. Once published, this version is prepared automatically on active hosts.

{#if canManage}
{/if} {:else if item.state === 'importing'}

The existing host import continues independently. It is reconciled when the host reconnects.

{/if} {/each} diff --git a/src/routes/hosts/+page.svelte b/src/routes/hosts/+page.svelte index 804829c..2a536b5 100644 --- a/src/routes/hosts/+page.svelte +++ b/src/routes/hosts/+page.svelte @@ -37,11 +37,11 @@ Hosts · Electric Cherry
- + {#if error}{/if} {#if notice}

{notice}

{/if} {#if canManage}
-

Add host

Run the installation command on a Debian 12 or 13 amd64 machine. Each host connects to this controller over HTTPS.

+

Add host

Run the installation command on a Debian 12 or 13 amd64 machine. Each host connects to this controller over HTTPS, then prepares its runtime and published game files automatically.

@@ -49,9 +49,9 @@
{#if enrollment}

Install the agent

This single-use command expires {new Date(enrollment.expiresAt).toLocaleString()}. Run it with administrator access on your game host.

{enrollment.command}
Open host setup →
{/if} -
{:else}

Administrators manage host setup, runtime preparation, and credentials.

{/if} +
{:else}

Administrators enroll hosts and manage their settings and credentials.

{/if} {#if loading}

Loading hosts…

{:else if !hosts.length}

No hosts connected yet.{canManage ? ' Add a host to begin.' : ''}

{:else} -
{#each hosts as host (host.id)}

{host.name}

{hostReadinessLabel(host.readiness)}

{host.hostname || 'Awaiting enrollment'} · {host.connectionState}

{host.inventory.os ?? 'OS pending'} · {host.inventory.architecture ?? 'Architecture pending'}

Game ports {host.portMin}–{host.portMax}

{#if host.enrolled}

Agent {host.inventory.agentVersion ?? 'unknown'} · Auto updates: {hostAgentUpdateLabel(host.inventory)}

{/if}{#if host.decommission}

{hostDeletionMessage(host)}

{#if host.decommission.error}

{host.decommission.error}

{/if}{:else if host.readinessReasons.length}

{host.readinessReasons[0]}

{/if}
{#if host.readiness === 'needs-game-files'}{/if}{#if canManage && (!host.decommission || host.decommission.retryable)}{/if}
{/each}
+
{#each hosts as host (host.id)}

{host.name}

{hostReadinessLabel(host.readiness)}

{host.hostname || 'Awaiting enrollment'} · {host.connectionState}

{host.inventory.os ?? 'OS pending'} · {host.inventory.architecture ?? 'Architecture pending'}

Game ports {host.portMin}–{host.portMax}

{#if host.enrolled}

Agent {host.inventory.agentVersion ?? 'unknown'} · Auto updates: {hostAgentUpdateLabel(host.inventory)}

{/if}{#if host.decommission}

{hostDeletionMessage(host)}

{#if host.decommission.error}

{host.decommission.error}

{/if}{:else}{#if host.preparation}

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

{#if host.preparation.state === 'blocked'}

{host.preparation.items.find(item => item.state === 'blocked')?.detail}

{/if}{/if}{#if host.readinessReasons.length}

{host.readinessReasons[0]}

{/if}{/if}
{#if canManage && (!host.decommission || host.decommission.retryable)}{/if}
{/each}
{/if}
diff --git a/src/routes/hosts/[id]/+page.svelte b/src/routes/hosts/[id]/+page.svelte index 16c2c27..658f64d 100644 --- a/src/routes/hosts/[id]/+page.svelte +++ b/src/routes/hosts/[id]/+page.svelte @@ -7,6 +7,7 @@ import SectionHeader from '$ui/components/primitives/SectionHeader.svelte'; import Button from '$ui/components/primitives/Button.svelte'; import DeleteHostDialog from '$ui/components/DeleteHostDialog.svelte'; + import HostPreparation from '$ui/components/HostPreparation.svelte'; let host = $state(null); let operations = $state([]); let loading = $state(true); let busy = $state(false); let error = $state(''); let notice = $state(''); let deleteOpen = $state(false); @@ -19,7 +20,7 @@ async function load(initialize = false) { try { const response = await fetch(endpoint); const body = await response.json(); if (response.status === 404 && host?.decommission) { await goto('/hosts'); return; } if (!response.ok) throw new Error(body.error ?? body.message ?? 'Could not load host.'); host = body.host; operations = body.operations; - if (initialize && host) { name = host.name; portMin = host.portMin; portMax = host.portMax; bindAddress = host.bindAddress; publicAddress = host.publicAddress; minimumFreeGB = host.minimumFreeBytes / 1073741824; } + if (initialize && host) { name = host.name; portMin = host.portMin; portMax = host.portMax; bindAddress = host.bindAddress; publicAddress = host.publicAddress; minimumFreeGB = host.minimumFreeBytes / 1073741824; runtimeImage = body.defaultRuntimeImage ?? runtimeImage; } } catch (cause) { error = (cause as Error).message; } finally { loading = false; } } async function action(path: string, method: string, body?: unknown, message?: string) { @@ -41,10 +42,10 @@ {#if host.decommission}

Host cleanup

{hostDeletionMessage(host)}

{#if host.decommission.error}

{host.decommission.error}

{/if}

Requested {new Date(host.decommission.requestedAt).toLocaleString()}. This page closes only after the cleanup worker confirms removal.

{#if host.decommission.stage === 'cleaned'}

The agent is expected to go offline during uninstall. If final confirmation never arrives, inspect the machine; loss of contact alone does not prove completion.

{/if}
{/if}

{host.connectionState} · {hostReadinessLabel(host.readiness)}

Last heartbeat: {host.lastHeartbeat ? new Date(host.lastHeartbeat).toLocaleString() : 'Never'}
{#if host.readinessReasons.length}
    {#each host.readinessReasons as reason}
  • {reason}
    • {/each}
{:else}

Connected and ready. Each deployment also checks its game library, runtime, and ports.

{/if} - {#if host.readiness === 'needs-game-files'}

Choose a published game-file version and download it to this host →

{/if} {#if canManage && !cleaning}
{/if}

Draining prevents new deployments and retires instances as they become safe to remove. Disabled hosts receive no new commands; running game containers continue.

+ {#if !cleaning}

Automatic host setup

{/if}

Inventory

Hostname
{host.hostname || 'Pending'}
Agent
{host.inventory.agentVersion ?? 'Unknown'}
Platform
{host.inventory.os ?? 'Unknown'} · {host.inventory.architecture ?? 'Unknown'}
CPU
{host.inventory.cpu ?? 'Unknown'} ({host.inventory.cpuCount ?? '?'} cores)
Memory available / total
{bytes(host.inventory.memoryAvailable)} / {bytes(host.inventory.memoryTotal)}
Game storage available / total
{bytes(host.inventory.storageAvailable)} / {bytes(host.inventory.storageTotal)}
Docker
{host.inventory.dockerVersion ?? 'Unknown'} · {host.inventory.dockerRunning ? 'Running' : 'Unavailable'}
Current operation
{host.inventory.currentOperation ?? 'None reported'}

Host configuration

{#if canManage}
@@ -57,8 +58,9 @@ {#if host.enrolled}

{hostAgentUpdateMessage(host.inventory)}

{:else}

The installer enables automatic updates when this host first enrolls.

{/if} {#if host.enrolled && !hostAgentUpdatesEnabled(host.inventory)}

{canManage ? 'Generate an installer under Agent access below, run it on this host, and wait for a fresh heartbeat. The installer preserves the existing host identity and game data.' : 'An administrator can generate an installer under Agent access and run it on this host.'}

{:else if hostAgentUpdatesEnabled(host.inventory)}

The host checks this controller about every five minutes. Updates replace agent code between operations; running game containers continue.

{/if}
-

Game files and runtime

Manage game-file libraries →

{#if host.inventory.libraries?.length}
    {#each host.inventory.libraries as library}
  • {library.id}: {library.state}{library.progress !== undefined ? ` (${library.progress}%)` : ''}
    • {/each}
{:else}

No cached game libraries reported.

{/if} - {#if canManage} { event.preventDefault(); void action('/operations', 'POST', { kind: 'runtime.pull', image: runtimeImage }, 'Runtime download queued.'); }}>
{/if}

Cached images: {host.inventory.runtimeImages?.join(', ') || 'None reported'}

+

Cached game files and runtimes

{#if host.inventory.libraries?.length}
    {#each host.inventory.libraries as library}
  • {library.id}: {library.state}{library.progress !== undefined ? ` (${library.progress}%)` : ''}
    • {/each}
{:else}

No cached game libraries reported yet.

{/if} +

Cached images: {host.inventory.runtimeImages?.join(', ') || 'None reported yet'}

+ {#if canManage}
Advanced: custom runtime or manual retry

The configured default runtime is prepared automatically. Use this to prepare another image or retry a pull now.

{ event.preventDefault(); void action('/operations', 'POST', { kind: 'runtime.pull', image: runtimeImage }, 'Runtime download queued.'); }}>
{/if}

Agent access

Credentials: {host.credentialActive ? 'Active' : 'Not active'}. Enrollment links expire after 15 minutes and work once.

{#if canManage}
{#if enrollment}

Expires {new Date(enrollment.expiresAt).toLocaleString()}

{enrollment.command}
{/if}{:else}

Administrators manage enrollment and agent credentials.

{/if}
diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index ff5517e..0ae344f 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -7,6 +7,7 @@ unknown and is reported for reconciliation, never automatically executed twice. import argparse import base64 +from concurrent.futures import ThreadPoolExecutor, as_completed import contextlib import hashlib import fcntl @@ -29,9 +30,10 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.2.0" +VERSION = "1.2.1" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 +LIBRARY_DOWNLOAD_WORKERS = 4 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") @@ -788,13 +790,46 @@ class Agent: pass libraries, caches = [], [] for marker in (self.root / "libraries").glob("*/*/manifest.json"): + # A crash between writing a manifest and its atomic promotion must + # never expose the staging directory as a ready library. + if marker.parent.name.startswith("."): + continue try: value = json.loads(marker.read_text()) libraries.append({"id": value["libraryId"], "state": "ready"}) caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}) except (OSError, ValueError, KeyError): continue - progress = list(self.library_progress.values()) + ready = {item["id"] for item in libraries} + progress = {key: value.copy() for key, value in list(self.library_progress.items()) if key not in ready} + for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): + try: + value = json.loads(marker.read_text()) + library_id = identifier(value["libraryId"]) + if library_id in ready or library_id in progress: + continue + expected = self.library_dir(value["game"], library_id).with_name("." + library_id + ".partial") + if marker.parent != expected or marker.is_symlink() or marker.parent.is_symlink(): + continue + files = value["files"] + canonical = json.dumps([{ "path": item["path"], "size": item["size"], "sha256": item["sha256"] } for item in files], separators=(",", ":"), ensure_ascii=False) + if digest(canonical.encode("utf-8")) != value["manifestSha256"]: + continue + retained, size, paths = 0, 0, set() + for item in files: + if item["path"] in paths or type(item["size"]) is not int or item["size"] < 0: + raise AgentError("Invalid partial library manifest") + paths.add(item["path"]) + target = scoped(marker.parent / "content", item["path"]) + size += item["size"] + if target.exists() and stat_module.S_ISREG(target.stat().st_mode): + retained += min(target.stat().st_size, item["size"]) + # These are resumable bytes, not verified content. Reporting + # them survives restarts without requiring space for two copies. + progress[library_id] = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesTotal": size, "manifestSha256": value["manifestSha256"]} + except (AgentError, OSError, ValueError, KeyError, TypeError): + continue + progress = list(progress.values()) libraries.extend({"id": item["libraryVersionId"], "state": item["status"]} for item in progress) caches.extend(progress) images = [] @@ -1195,6 +1230,65 @@ class Agent: raise AgentError("Artifact redirects are not allowed") return urllib.request.build_opener(NoRedirect()).open(req, timeout=60) + def download_library_file(self, item, target, files_url, update_progress, aborted): + def check_active(): + if aborted.is_set(): + raise AgentError("Library download stopped after a transfer failed") + + check_active() + target.parent.mkdir(parents=True, exist_ok=True) + offset = target.stat().st_size if target.exists() else 0 + if offset > item["size"]: + target.unlink() + offset = 0 + checksum = hashlib.sha256() + # Hash the saved prefix once so resumed transfers still verify the whole + # file. New bytes are hashed as they arrive, avoiding a second disk read. + if offset: + with target.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + check_active() + checksum.update(chunk) + if offset == item["size"] and target.exists(): + if checksum.hexdigest() == item["sha256"]: + update_progress(item["path"], offset) + return + target.unlink() + offset = 0 + checksum = hashlib.sha256() + update_progress(item["path"], offset) + check_active() + url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") + with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response: + if response.status == 206: + expected = f"bytes {offset}-{item['size'] - 1}/{item['size']}" + if response.headers.get("Content-Range") != expected: + raise AgentError("Invalid resumed download response") + elif response.status != 200: + raise AgentError("Invalid library download response") + elif offset: + # A server may ignore Range and send the whole file instead. + offset = 0 + checksum = hashlib.sha256() + update_progress(item["path"], 0) + with target.open("ab" if offset else "wb") as output: + received = offset + while True: + check_active() + chunk = response.read(1024 * 1024) + if not chunk: + break + received += len(chunk) + if received > item["size"]: + raise AgentError("Library download exceeded its manifest size") + output.write(chunk) + checksum.update(chunk) + update_progress(item["path"], received) + output.flush() + os.fsync(output.fileno()) + if target.stat().st_size != item["size"] or checksum.hexdigest() != item["sha256"]: + raise AgentError("Downloaded library file failed integrity verification") + def library_download(self, payload): library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId"))) with self.artifact_open("GET", payload["manifestUrl"]) as response: @@ -1221,55 +1315,46 @@ class Agent: source_marker = staging / "source.json" if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha: raise Rejected("Partial library download belongs to a different manifest") - atomic_json(source_marker, {"libraryId": library_id, "manifestSha256": manifest_sha}) size = sum(item["size"] for item in files) - downloaded = 0 - paths = set() + atomic_json(source_marker, {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha}) + targets = {} progress = {"libraryVersionId": library_id, "status": "downloading", "bytesDownloaded": 0, "bytesTotal": size, "manifestSha256": manifest_sha} self.library_progress[library_id] = progress try: if shutil.disk_usage(staging).free < max(0, size - sum(path.stat().st_size for path in staging.rglob("*") if path.is_file())): raise AgentError("Insufficient library storage") + # Validate every path before starting concurrent writes. In addition + # to duplicates, a manifest cannot treat a file as a parent folder. for item in files: - if item["path"] in paths or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): + if item["path"] in targets or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): raise AgentError("Invalid library file manifest") - paths.add(item["path"]) - target = scoped(staging / "content", item["path"]) - target.parent.mkdir(parents=True, exist_ok=True) - offset = target.stat().st_size if target.exists() else 0 - if offset > item["size"]: - target.unlink() - offset = 0 - if offset == item["size"] and target.exists() and file_digest(target) != item["sha256"]: - target.unlink() - offset = 0 - if offset < item["size"] or not target.exists(): - url = payload["filesBaseUrl"].rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") - with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response: - if offset and response.status == 206: - content_range = response.headers.get("Content-Range", "") - if not content_range.startswith(f"bytes {offset}-"): - raise AgentError("Invalid resumed download response") - elif offset: - offset = 0 - with target.open("ab" if offset else "wb") as output: - received = offset - while True: - chunk = response.read(1024 * 1024) - if not chunk: - break - received += len(chunk) - if received > item["size"]: - raise AgentError("Library download exceeded its manifest size") - output.write(chunk) - progress["bytesDownloaded"] = downloaded + received - output.flush() - os.fsync(output.fileno()) - progress["status"] = "verifying" - if target.stat().st_size != item["size"] or file_digest(target) != item["sha256"]: - raise AgentError("Downloaded library file failed integrity verification") - downloaded += item["size"] - progress.update(status="downloading", bytesDownloaded=downloaded) + targets[item["path"]] = scoped(staging / "content", item["path"]) + for path in targets: + if any(str(parent) in targets for parent in PurePosixPath(path).parents): + raise AgentError("Library file paths overlap") + progress_lock = threading.Lock() + offsets = {} + aborted = threading.Event() + + def update_progress(path, offset): + with progress_lock: + progress["bytesDownloaded"] += offset - offsets.get(path, 0) + offsets[path] = offset + + def transfer(item): + try: + self.download_library_file(item, targets[item["path"]], payload["filesBaseUrl"], update_progress, aborted) + except Exception: + aborted.set() + raise + + # Bound sockets, memory and disk pressure while allowing other files + # to progress during network round trips and per-file fsyncs. + with ThreadPoolExecutor(max_workers=min(LIBRARY_DOWNLOAD_WORKERS, len(files))) as pool: + futures = [pool.submit(transfer, item) for item in files] + for future in as_completed(futures): + future.result() + progress["status"] = "verifying" saved = {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha} atomic_json(staging / "manifest.json", saved) for path in (staging / "content").rglob("*"): @@ -1446,6 +1531,16 @@ 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": + # 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. + self.journal.update(operation["id"], "running") + try: + result = self.dispatch(operation) + return self.journal.update(operation["id"], "completed", {**result, "reconciled": True}) + except Exception: + return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed") if kind == "library.import": try: with self.artifact_open("GET", payload["completeUrl"]) as response: diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 094e349..57935aa 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.2.0", + "version": "1.2.1", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "c0c08bc50daf09f4e6e141d4119cee68973b47c29845b2f3620e4579d9cd7daa", + "sha256": "680c0c2bd6d703d3b0119ff5b253f1f6b70d1eeee98fe0e6770738eab2b6d380", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/host-preparation.test.ts b/tests/unit/host-preparation.test.ts new file mode 100644 index 0000000..15cec21 --- /dev/null +++ b/tests/unit/host-preparation.test.ts @@ -0,0 +1,357 @@ +import { createHash } from 'node:crypto'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { GameLibraryVersion } from '../../src/lib/game-files.js'; +import type { HostInventory } from '../../src/lib/hosts.js'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +let fixture: Fixture; +let hosts: typeof import('../../src/lib/server/hosts/index.js'); +let libraries: typeof import('../../src/lib/server/game-files/index.js'); +let preparation: typeof import('../../src/lib/server/hosts/preparation.js'); +const runtime = 'ghcr.io/example/plutainer:provisioning-test'; +const initialTime = Date.parse('2026-10-04T12:00:00Z'); +const content = 'a'.repeat(100); +const inventory: HostInventory = { + hostname: 'game-host', protocolVersion: 1, dockerRunning: true, directoriesWritable: true, + storageAvailable: 100 * 1024 ** 3, runtimeImages: [], libraries: [], capabilities: ['host.decommission'] +}; + +beforeEach(async () => { + vi.resetModules(); + fixture = await newFixture([]); + vi.stubEnv('PLUTAINER_IMAGE', runtime); + vi.stubEnv('GAME_FILES_ROOT', join(fixture.root, 'game-files')); + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(initialTime); + hosts = await import('../../src/lib/server/hosts/index.js'); + libraries = await import('../../src/lib/server/game-files/index.js'); + preparation = await import('../../src/lib/server/hosts/preparation.js'); +}); + +afterEach(async () => { + vi.useRealTimers(); + vi.unstubAllEnvs(); + await fixture.cleanup(); +}); + +function enroll(overrides: HostInventory = {}) { + const host = hosts.createHost({ name: 'Automatic game host' }); + hosts.enrollHost(hosts.createEnrollment(host.id).token, { ...inventory, ...overrides }); + return host.id; +} + +function heartbeat(id: string, overrides: HostInventory = {}) { + vi.setSystemTime(Date.now() + 1); + return hosts.heartbeat(id, { ...inventory, ...overrides }); +} + +function operations(id: string) { + return hosts.listHostOperations(id).filter(operation => ['runtime.pull', 'library.download'].includes(operation.kind)); +} + +async function publish(name = 'T5 files', version = '1', game: 't5' | 't6' = 't5', bytes = content) { + vi.setSystemTime(Date.now() + 1); + const upload = libraries.createLibraryUpload({ + name, version, game, files: [{ path: 'zone/game.ff', size: bytes.length, lastModified: 0 }] + }); + await libraries.uploadBrowserFile(upload.id, 'zone/game.ff', { + offset: 0, size: bytes.length, sha256: createHash('sha256').update(bytes).digest('hex') + }, new Response(bytes).body); + return libraries.publishBrowserUpload(upload.id); +} + +function verified(id: string, version: GameLibraryVersion) { + libraries.updateHostLibraryCaches(id, [{ + libraryVersionId: version.id, status: 'ready', manifestSha256: version.manifestSha256 + }]); +} + +describe('automatic host preparation', () => { + it('prepares an enrolled host through authenticated agent polls and heartbeats without manual setup requests', async () => { + const [{ POST: enrollAgent }, { GET: pollOperations }, { POST: reportOperation }, { POST: reportHeartbeat }] = await Promise.all([ + import('../../src/routes/api/agent/enroll/+server.js'), + import('../../src/routes/api/agent/operations/+server.js'), + import('../../src/routes/api/agent/operations/[id]/+server.js'), + import('../../src/routes/api/agent/heartbeat/+server.js') + ]); + const host = hosts.createHost({ name: 'Agent provisioning integration' }); + const enrollment = await enrollAgent({ request: new Request('https://controller.test/api/agent/enroll', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ token: hosts.createEnrollment(host.id).token, inventory }) + }) }); + expect(enrollment.status).toBe(200); + const { credential } = await enrollment.json(); + const request = (path: string, body?: object) => new Request(`https://controller.test/api/agent/${path}`, { + method: body ? 'POST' : 'GET', + headers: { Authorization: `Bearer ${credential}`, 'Content-Type': 'application/json' }, + ...(body ? { body: JSON.stringify(body) } : {}) + }); + expect(operations(host.id)).toEqual([]); + + const firstPoll = await pollOperations({ request: request('operations') }).json(); + expect(firstPoll.operations).toHaveLength(1); + const runtimeOperation = firstPoll.operations[0]; + expect(runtimeOperation).toMatchObject({ kind: 'runtime.pull', payload: { image: runtime } }); + expect((await pollOperations({ request: request('operations') }).json()).operations).toEqual([runtimeOperation]); + const version = await publish(); + expect((await pollOperations({ request: request('operations') }).json()).operations).toEqual([runtimeOperation]); + + vi.setSystemTime(Date.now() + 1); + const completedRuntime = await reportOperation({ + request: request(`operations/${runtimeOperation.id}`, { state: 'completed' }), params: { id: runtimeOperation.id } + }); + expect(completedRuntime.status).toBe(200); + expect((await pollOperations({ request: request('operations') }).json()).operations).toEqual([]); + expect(operations(host.id)).toHaveLength(1); + + vi.setSystemTime(Date.now() + 1); + const preparedInventory = { ...inventory, runtimeImages: [runtime], libraryCaches: [] }; + const runtimeHeartbeat = await reportHeartbeat({ request: request('heartbeat', { inventory: preparedInventory }) }); + expect(runtimeHeartbeat.status).toBe(200); + expect(operations(host.id)).toHaveLength(2); + const download = operations(host.id)[0]; + expect(download).toMatchObject({ kind: 'library.download', payload: { libraryVersionId: version.id } }); + expect((await reportHeartbeat({ request: request('heartbeat', { inventory: preparedInventory }) })).status).toBe(200); + expect((await pollOperations({ request: request('operations') }).json()).operations).toEqual([download]); + expect(operations(host.id)).toHaveLength(2); + + vi.setSystemTime(Date.now() + 1); + expect((await reportOperation({ + request: request(`operations/${download.id}`, { state: 'completed' }), params: { id: download.id } + })).status).toBe(200); + vi.setSystemTime(Date.now() + 1); + const finalHeartbeat = await reportHeartbeat({ request: request('heartbeat', { inventory: { + ...preparedInventory, + libraries: [{ id: version.id, state: 'ready' }], + libraryCaches: [{ libraryVersionId: version.id, status: 'ready', manifestSha256: version.manifestSha256 }] + } }) }); + expect(finalHeartbeat.status).toBe(200); + expect(libraries.requireReadyLibrary(host.id, version.id).id).toBe(version.id); + expect(preparation.hostPreparationStatus(hosts.getHost(host.id)!).state).toBe('ready'); + expect((await pollOperations({ request: request('operations') }).json()).operations).toEqual([]); + expect(operations(host.id)).toHaveLength(2); + }); + + it('presents preparation without queuing work and prepares the configured runtime even when another image exists', async () => { + const version = await publish(); + const id = enroll({ runtimeImages: ['another:runtime'] }); + const host = hosts.getHost(id)!; + const status = preparation.hostPreparationStatus(host); + expect(status.items.map(item => item.kind)).toEqual(['runtime', 'library']); + expect(preparation.presentHostPreparation(host)).toMatchObject({ id, preparation: status }); + expect(operations(id)).toEqual([]); + + expect(preparation.reconcileHostPreparation(id).state).toBe('preparing'); + const [operation] = operations(id); + expect(operation).toMatchObject({ kind: 'runtime.pull', payload: { image: runtime }, state: 'pending' }); + expect(libraries.getHostLibraryCache(id, version.id)).toBeNull(); + for (let poll = 0; poll < 4; poll++) { + heartbeat(id); + preparation.reconcileHostPreparation(id); + } + hosts.reportHostOperation(id, operation.id, { state: 'unknown' }); + preparation.reconcileHostPreparation(id); + expect(operations(id).map(item => item.id)).toEqual([operation.id]); + }); + + it('waits for fresh inventory after runtime completion before starting a library download', async () => { + const version = await publish(); + const id = enroll(); + preparation.reconcileHostPreparation(id); + const runtimeOperation = operations(id)[0]; + vi.setSystemTime(Date.now() + 1); + hosts.reportHostOperation(id, runtimeOperation.id, { state: 'completed', result: { image: runtime } }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + + heartbeat(id, { runtimeImages: [runtime] }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(2); + const download = operations(id)[0]; + expect(download).toMatchObject({ kind: 'library.download', payload: { libraryVersionId: version.id } }); + verified(id, version); + hosts.reportHostOperation(id, download.id, { state: 'completed' }); + heartbeat(id, { runtimeImages: [runtime], libraries: [{ id: version.id, state: 'ready' }] }); + expect(preparation.reconcileHostPreparation(id).state).toBe('ready'); + expect(operations(id)).toHaveLength(2); + }); + + it('downloads the newest published version of every library, including separate libraries for the same game', async () => { + const old = await publish('T5 files', 'old'); + const latest = await publish('T5 files', 'new'); + const alternate = await publish('T5 alternate'); + const t6 = await publish('T6 files', '1', 't6'); + libraries.createLibraryUpload({ + name: 'T5 files', version: 'unfinished', game: 't5', + files: [{ path: 'zone/game.ff', size: content.length, lastModified: 0 }] + }); + const id = enroll({ runtimeImages: [runtime] }); + const expected = new Map([latest, alternate, t6].map(version => [version.id, version])); + const downloaded: string[] = []; + for (let index = 0; index < expected.size; index++) { + preparation.reconcileHostPreparation(id); + const active = operations(id).filter(operation => !['completed', 'failed'].includes(operation.state)); + expect(active).toHaveLength(1); + const operation = active[0]; + const versionId = String(operation.payload.libraryVersionId); + expect(expected.has(versionId)).toBe(true); + expect(downloaded).not.toContain(versionId); + downloaded.push(versionId); + verified(id, expected.get(versionId)!); + vi.setSystemTime(Date.now() + 1); + hosts.reportHostOperation(id, operation.id, { state: 'completed' }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(index + 1); + heartbeat(id, { runtimeImages: [runtime], libraries: downloaded.map(id => ({ id, state: 'ready' })) }); + } + expect(preparation.reconcileHostPreparation(id).state).toBe('ready'); + expect(downloaded).toHaveLength(expected.size); + expect(downloaded).not.toContain(old.id); + }); + + it('picks up publication after enrollment and keeps verified historical versions intact', async () => { + const id = enroll({ runtimeImages: [runtime] }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toEqual([]); + const old = await publish(); + verified(id, old); + heartbeat(id, { runtimeImages: [runtime], libraries: [{ id: old.id, state: 'ready' }] }); + expect(preparation.reconcileHostPreparation(id).state).toBe('ready'); + const latest = await publish('T5 files', '2'); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + expect(operations(id)[0].payload.libraryVersionId).toBe(latest.id); + expect(libraries.requireReadyLibrary(id, old.id).id).toBe(old.id); + expect(libraries.getHostLibraryCache(id, old.id)?.status).toBe('ready'); + }); + + it.each(['draining', 'disabled'] as const)('does not start setup for a %s host', mode => { + const id = enroll(); + hosts.updateHost(id, { mode }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toEqual([]); + }); + + it.each([ + { name: 'incompatible protocol', value: { protocolVersion: 2 } }, + { name: 'unavailable Docker', value: { dockerRunning: false } }, + { name: 'unwritable managed directories', value: { directoriesWritable: false } }, + { name: 'insufficient storage reserve', value: { storageAvailable: 0 } } + ])('does not start setup with $name', ({ value }) => { + const id = enroll(value); + expect(preparation.reconcileHostPreparation(id).state).toBe('blocked'); + expect(operations(id)).toEqual([]); + }); + + it('does not start setup for unenrolled, disconnected, revoked, or decommissioning hosts', () => { + const unenrolled = hosts.createHost({ name: 'Not enrolled' }).id; + preparation.reconcileHostPreparation(unenrolled); + expect(operations(unenrolled)).toEqual([]); + const offline = enroll(); + vi.setSystemTime(Date.now() + 91_000); + preparation.reconcileHostPreparation(offline); + expect(operations(offline)).toEqual([]); + const revoked = enroll(); + hosts.revokeCredentials(revoked); + preparation.reconcileHostPreparation(revoked); + expect(operations(revoked)).toEqual([]); + const deleting = enroll(); + hosts.deleteHost(deleting); + preparation.reconcileHostPreparation(deleting); + expect(operations(deleting)).toEqual([]); + }); + + it('retries runtime failures with exponential delays capped at fifteen minutes', () => { + const id = enroll(); + preparation.reconcileHostPreparation(id); + for (const [attempt, delay] of [60_000, 120_000, 240_000, 480_000, 900_000, 900_000].entries()) { + const operation = operations(id)[0]; + vi.setSystemTime(Date.now() + 1); + hosts.reportHostOperation(id, operation.id, { state: 'failed', error: 'Registry unavailable' }); + const failedAt = Date.now(); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(attempt + 1); + vi.setSystemTime(failedAt + delay - 2); + heartbeat(id); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(attempt + 1); + vi.setSystemTime(failedAt + delay); + heartbeat(id); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(attempt + 2); + expect(operations(id)[0]).toMatchObject({ kind: 'runtime.pull', state: 'pending' }); + } + }); + + it('blocks a library that cannot fit while preserving the host free-space reserve', async () => { + const version = await publish(); + const id = enroll({ 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([]); + heartbeat(id, { runtimeImages: [runtime], storageAvailable: 1_000 + version.sizeBytes }); + expect(preparation.reconcileHostPreparation(id).state).toBe('preparing'); + expect(operations(id)[0].payload.libraryVersionId).toBe(version.id); + }); + + it('returns an active download even when its in-progress bytes have consumed free space', async () => { + const version = await publish(); + const id = enroll({ runtimeImages: [runtime] }); + const operation = libraries.queueLibraryDownload(id, version.id); + heartbeat(id, { runtimeImages: [runtime], storageAvailable: hosts.getHost(id)!.minimumFreeBytes }); + expect(libraries.queueLibraryDownload(id, version.id).id).toBe(operation.id); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + }); + + it('lets other libraries prepare when an earlier library cannot fit', async () => { + const oversized = await publish('A oversized'); + const small = await publish('B small', '1', 't6', 'small'); + const id = enroll({ runtimeImages: [runtime], storageAvailable: 1_010 }); + hosts.updateHost(id, { minimumFreeBytes: 1_000 }); + const status = preparation.reconcileHostPreparation(id); + expect(status.items).toEqual(expect.arrayContaining([ + expect.objectContaining({ label: 'A oversized · 1', state: 'blocked' }), + expect.objectContaining({ label: 'B small · 1', state: 'preparing' }) + ])); + expect(operations(id)).toHaveLength(1); + expect(operations(id)[0].payload.libraryVersionId).toBe(small.id); + expect(libraries.getHostLibraryCache(id, oversized.id)).toBeNull(); + }); + + it('does not count partial bytes from a different manifest toward download storage', async () => { + const version = await publish(); + const id = enroll({ runtimeImages: [runtime] }); + libraries.updateHostLibraryCaches(id, [{ + libraryVersionId: version.id, status: 'failed', bytesDownloaded: 80, manifestSha256: '0'.repeat(64) + }]); + heartbeat(id, { runtimeImages: [runtime], storageAvailable: hosts.getHost(id)!.minimumFreeBytes + 20 }); + expect(preparation.reconcileHostPreparation(id).state).toBe('blocked'); + expect(() => libraries.queueLibraryDownload(id, version.id)).toThrow(/insufficient storage/); + expect(operations(id)).toEqual([]); + }); + + it('resumes a failed verified-manifest download using only its remaining storage', async () => { + const version = await publish(); + const id = enroll({ runtimeImages: [runtime] }); + preparation.reconcileHostPreparation(id); + const operation = operations(id)[0]; + libraries.updateHostLibraryCaches(id, [{ + libraryVersionId: version.id, status: 'downloading', bytesDownloaded: 80, manifestSha256: version.manifestSha256 + }]); + vi.setSystemTime(Date.now() + 1); + hosts.reportHostOperation(id, operation.id, { state: 'failed', error: 'Connection reset' }); + const failedAt = Date.now(); + const free = hosts.getHost(id)!.minimumFreeBytes + version.sizeBytes - 80; + heartbeat(id, { runtimeImages: [runtime], storageAvailable: free }); + preparation.reconcileHostPreparation(id); + expect(operations(id)).toHaveLength(1); + vi.setSystemTime(failedAt + 60_000); + heartbeat(id, { runtimeImages: [runtime], storageAvailable: free }); + expect(preparation.reconcileHostPreparation(id).state).toBe('preparing'); + expect(operations(id)).toHaveLength(2); + expect(operations(id)[0].payload.libraryVersionId).toBe(version.id); + expect(libraries.getHostLibraryCache(id, version.id)?.bytesDownloaded).toBe(80); + }); +});