From f8c60fd101dc79a11983f607db3649b9c59eb5c3 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Sun, 4 Oct 2026 12:51:05 -0400 Subject: [PATCH] Separate game file downloads and verification with progress timing --- README.md | 7 +- agent/ec_agent.py | 248 ++++++++++++---- agent/tests/test_agent.py | 272 +++++++++++++++++- docs/native-hosts.md | 22 +- src/lib/game-files.ts | 6 + src/lib/hosts.ts | 3 + src/lib/server/game-files/index.ts | 54 +++- src/lib/server/game-files/schema.ts | 3 + src/lib/server/hosts/preparation.ts | 7 +- .../GameFileTransferProgress.svelte | 52 ++++ src/lib/ui/components/HostPreparation.svelte | 6 +- src/lib/ui/game-file-transfer.ts | 33 +++ src/routes/game-files/+page.svelte | 7 +- static/agent/ec-agent.py | 248 ++++++++++++---- static/agent/manifest.json | 4 +- .../unit/game-file-transfer-progress.test.ts | 236 +++++++++++++++ tests/unit/ui-game-file-transfer.test.ts | 74 +++++ 17 files changed, 1150 insertions(+), 132 deletions(-) create mode 100644 src/lib/ui/components/GameFileTransferProgress.svelte create mode 100644 src/lib/ui/game-file-transfer.ts create mode 100644 tests/unit/game-file-transfer-progress.test.ts create mode 100644 tests/unit/ui-game-file-transfer.test.ts diff --git a/README.md b/README.md index aebdb5c..4ab4c82 100644 --- a/README.md +++ b/README.md @@ -89,9 +89,10 @@ publication. Actual game files use dedicated filesystem storage, never SQLite blobs. The last upload to finish verification becomes that game's latest files, even if it started earlier. Active hosts always prepare the latest T5 and T6 files when available, 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 +hosts enrolled later. Hosts first download all files with up to four concurrent +transfers, then verify all checksums in a separate pass before the files become +usable. Both steps show progress, elapsed time, measured speed and an estimated +time remaining for the current step. Interrupted transfers resume. 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 existing files and runtime without a forced restart. diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 0ae344f..22f605e 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -30,7 +30,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.2.1" +VERSION = "1.2.2" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 LIBRARY_DOWNLOAD_WORKERS = 4 @@ -587,6 +587,80 @@ class Uninstaller: time.sleep(15) +class LibraryTransfer: + """Durable phase progress; throughput counts only work in this attempt.""" + + def __init__(self, progress, marker, previous=None): + self.progress = progress + self.marker = marker + self.lock = threading.RLock() + prior = (previous or {}).get("transfer") or {} + self.prior_phase = prior.get("phase") + self.prior_phase_elapsed = prior.get("phaseElapsedSeconds", 0) + self.started_at = prior.get("startedAt") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + self.elapsed = prior.get("elapsedSeconds", 0) + self.phase_elapsed = prior.get("phaseElapsedSeconds", 0) if prior.get("phase") == progress["status"] else 0 + self.started = self.phase_started = time.monotonic() + self.phase = progress["status"] + self.work = 0 + self.finished = False + self.failed_bytes_downloaded = None + self.persisted_at = self.started + self.persist(force=True) + + def snapshot(self): + with self.lock: + if not self.finished: + now = time.monotonic() + sample_seconds = max(0, now - self.phase_started) + rate = self.work / sample_seconds if sample_seconds >= 1 and self.work else None + done = self.progress["bytesDownloaded"] if self.phase == "downloading" else self.progress["bytesVerified"] + self.progress["transfer"] = { + "phase": self.phase, "startedAt": self.started_at, + "elapsedSeconds": self.elapsed + max(0, now - self.started), + "phaseElapsedSeconds": self.phase_elapsed + sample_seconds, + "bytesPerSecond": rate, + "etaSeconds": max(0, self.progress["bytesTotal"] - done) / rate if rate else None, + } + return {**self.progress, "transfer": self.progress["transfer"].copy()} + + def persist(self, force=False): + with self.lock: + now = time.monotonic() + self.snapshot() + if force or now - self.persisted_at >= 5: + atomic_json(self.marker, self.progress) + self.persisted_at = now + + def advance(self, key, count, work): + with self.lock: + self.progress[key] += count + self.work += work + self.persist() + + def begin_verification(self): + with self.lock: + self.progress.update(status="verifying", bytesVerified=0) + if self.phase != "verifying": + self.phase_elapsed = self.prior_phase_elapsed if self.prior_phase == "verifying" else 0 + self.phase = "verifying" + self.phase_started = time.monotonic() + self.work = 0 + self.persist(force=True) + + def finish(self, status, error=None): + with self.lock: + self.snapshot() + self.finished = True + self.progress["status"] = status + if status == "failed" and self.failed_bytes_downloaded is not None: + self.progress["bytesDownloaded"] = self.failed_bytes_downloaded + self.progress["transfer"]["etaSeconds"] = 0 if status == "ready" else None + if error: + self.progress["error"] = error + self.persist(force=True) + + class Agent: def __init__(self, config, docker=None): self.config = config @@ -599,6 +673,7 @@ class Agent: self.stopping = threading.Event() self.current_operation = None self.library_progress = {} + self.library_transfers = {} self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None @@ -789,6 +864,8 @@ class Agent: except (AgentError, ValueError): pass libraries, caches = [], [] + progress = {key: value.copy() for key, value in list(self.library_progress.items())} + progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())}) 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. @@ -796,12 +873,22 @@ class Agent: continue try: value = json.loads(marker.read_text()) + # Rechecking an existing cache must revoke readiness until its + # entire contents have passed verification again. + if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists(): + saved_progress = json.loads((marker.parent / "transfer.json").read_text()) + if saved_progress.get("status") != "ready": + progress[value["libraryId"]] = {**saved_progress, "status": "failed" if saved_progress.get("status") == "failed" else "missing"} + if value["libraryId"] in progress: + continue libraries.append({"id": value["libraryId"], "state": "ready"}) - caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}) + cache = {"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesVerified": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")} + if value.get("transfer"): + cache["transfer"] = value["transfer"] + caches.append(cache) except (OSError, ValueError, KeyError): continue 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()) @@ -826,7 +913,15 @@ class Agent: 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"]} + cache = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": value["manifestSha256"]} + transfer_marker = marker.parent / "transfer.json" + if transfer_marker.exists(): + saved_progress = json.loads(transfer_marker.read_text()) + if saved_progress.get("manifestSha256") == value["manifestSha256"]: + cache.update(bytesVerified=saved_progress.get("bytesVerified", 0), transfer=saved_progress.get("transfer")) + if saved_progress.get("status") == "failed": + cache.update(status="failed", error=saved_progress.get("error")) + progress[library_id] = cache except (AgentError, OSError, ValueError, KeyError, TypeError): continue progress = list(progress.values()) @@ -1221,14 +1316,14 @@ class Agent: raise AgentError("Library transport must use HTTPS") return url - def artifact_open(self, method, path, data=None, extra_headers=None): + def artifact_open(self, method, path, data=None, extra_headers=None, timeout=60): headers = {"Authorization": "Bearer " + self.config["credential"], **(extra_headers or {})} req = urllib.request.Request(self.artifact_url(path), data=data, method=method, headers=headers) # Redirects cannot forward host credentials to another origin. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, hdrs, newurl): raise AgentError("Artifact redirects are not allowed") - return urllib.request.build_opener(NoRedirect()).open(req, timeout=60) + return urllib.request.build_opener(NoRedirect()).open(req, timeout=timeout) def download_library_file(self, item, target, files_url, update_progress, aborted): def check_active(): @@ -1241,21 +1336,11 @@ class Agent: 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) + # Saved prefixes and complete files are intentionally not read here. + # The second phase verifies every byte after all transfers have ended. 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) + return update_progress(item["path"], offset) check_active() url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") @@ -1269,7 +1354,6 @@ class Agent: 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 @@ -1282,13 +1366,48 @@ class Agent: 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"]: + if target.stat().st_size != item["size"]: + raise AgentError("Downloaded library file size failed integrity verification") + + def verify_downloaded_files(self, files, targets, transfer, repair=False): + invalid = False + retained = transfer.progress["bytesDownloaded"] + for item in files: + target = targets[item["path"]] + checksum = hashlib.sha256() + actual_size = target.stat().st_size if target.is_file() else 0 + valid_size = target.is_file() and actual_size == item["size"] + if valid_size: + with target.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + checksum.update(chunk) + transfer.advance("bytesVerified", len(chunk), len(chunk)) + if not valid_size or checksum.hexdigest() != item["sha256"]: + invalid = True + if repair and target.exists(): + # A corrupt complete file (including a corrupt saved + # prefix) must not be resumed indefinitely on every retry. + target.unlink() + actual_size = 0 + retained -= item["size"] - min(actual_size, item["size"]) + if invalid: + # Verification describes a completed download phase. Keep that + # total stable while other files are checked, then publish retained + # bytes atomically with failure (including removed corrupt files). + transfer.failed_bytes_downloaded = retained raise AgentError("Downloaded library file failed integrity verification") + def publish_library_progress(self, payload, transfer): + if payload.get("statusUrl"): + # Phase boundaries are useful even when shorter than the heartbeat + # interval. Telemetry failure must never undo a verified download. + with contextlib.suppress(AgentError, OSError, ValueError): + with self.artifact_open("POST", payload["statusUrl"], json.dumps(transfer.snapshot()).encode(), {"Content-Type": "application/json"}, timeout=5) as response: + response.read() + def library_download(self, payload): library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId"))) with self.artifact_open("GET", payload["manifestUrl"]) as response: @@ -1305,12 +1424,12 @@ class Agent: if payload.get("manifestSha256") and payload["manifestSha256"] != manifest_sha: raise AgentError("Library manifest differs from the requested immutable version") directory = self.library_dir(manifest["game"], library_id) - if directory.exists(): - saved = self.verify_library(directory) + existing = directory.exists() + if existing: + saved = json.loads((directory / "manifest.json").read_text()) if saved.get("manifestSha256") != manifest_sha: raise AgentError("Published library changed unexpectedly") - return {"libraryId": library_id, "state": "ready", "size": saved["size"]} - staging = directory.with_name("." + library_id + ".partial") + staging = directory if existing else directory.with_name("." + library_id + ".partial") staging.mkdir(parents=True, exist_ok=True) source_marker = staging / "source.json" if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha: @@ -1318,28 +1437,39 @@ class Agent: size = sum(item["size"] for item in files) 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} + progress = {"libraryVersionId": library_id, "status": "verifying" if existing else "downloading", "bytesDownloaded": size if existing else 0, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": manifest_sha} self.library_progress[library_id] = progress + marker = staging / "transfer.json" + previous = json.loads(marker.read_text()) if marker.exists() else None + if previous and (previous.get("manifestSha256") != manifest_sha or previous.get("status") == "ready"): + previous = None + timing = LibraryTransfer(progress, marker, previous) + self.library_transfers[library_id] = timing 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 targets 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 type(item["size"]) is not int or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): raise AgentError("Invalid library file manifest") 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 = {} + offsets = {item["path"]: min(targets[item["path"]].stat().st_size, item["size"]) if targets[item["path"]].exists() else 0 for item in files} + if not existing: + progress["bytesDownloaded"] = sum(offsets.values()) + if not existing and shutil.disk_usage(staging).free < size - progress["bytesDownloaded"]: + raise AgentError("Insufficient library storage") + timing.persist(force=True) + self.publish_library_progress(payload, timing) aborted = threading.Event() def update_progress(path, offset): with progress_lock: - progress["bytesDownloaded"] += offset - offsets.get(path, 0) + delta = offset - offsets.get(path, 0) offsets[path] = offset + timing.advance("bytesDownloaded", delta, max(0, delta)) def transfer(item): try: @@ -1350,21 +1480,44 @@ class Agent: # 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" + if not existing: + # A crash after setting read-only permissions but before the + # atomic rename leaves a valid, resumable staging directory. + content_root = staging / "content" + if content_root.exists(): + os.chmod(content_root, 0o755) + for path in content_root.rglob("*"): + if path.is_symlink(): + raise AgentError("Symlinks are not allowed in library content") + os.chmod(path, 0o755 if path.is_dir() else 0o644) + 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() + timing.begin_verification() + self.publish_library_progress(payload, timing) + self.verify_downloaded_files(files, targets, timing, repair=not existing) 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("*"): - os.chmod(path, 0o555 if path.is_dir() else 0o444) - os.chmod(staging / "content", 0o555) - os.replace(staging, directory) + if not existing: + for path in (staging / "content").rglob("*"): + if path.is_symlink(): + raise AgentError("Symlinks are not allowed in library content") + os.chmod(path, 0o555 if path.is_dir() else 0o444) + os.chmod(staging / "content", 0o555) + os.replace(staging, directory) + timing.marker = directory / "transfer.json" + timing.finish("ready") + saved["transfer"] = timing.snapshot()["transfer"] + saved["bytesVerified"] = size + atomic_json(directory / "manifest.json", saved) + self.publish_library_progress(payload, timing) self.library_progress.pop(library_id, None) + self.library_transfers.pop(library_id, None) return {"libraryId": library_id, "state": "ready", "size": size, "manifestSha256": manifest_sha} except Exception: - progress.update(status="failed", error="Library download or integrity verification failed") + timing.finish("failed", "Library download or integrity verification failed") + self.publish_library_progress(payload, timing) raise def library_import(self, payload): @@ -1561,14 +1714,9 @@ class Agent: finally: self.executing_id = None if kind == "library.download": - library_id = payload.get("libraryVersionId", payload.get("libraryId")) - for game in ("t5", "t6"): - directory = self.library_dir(game, library_id) - if directory.exists(): - saved = self.verify_library(directory) - return self.journal.update(operation["id"], "completed", {"libraryId": library_id, "state": "ready", "size": saved["size"], "manifestSha256": saved.get("manifestSha256"), "reconciled": True}) # This is a resumable immutable transfer, not a repeated - # container command: verified completed files remain untouched. + # container command. Full files skip transfer; verification + # always starts again so interrupted reads cannot prove ready. self.journal.update(operation["id"], "running") try: return self.journal.update(operation["id"], "completed", self.library_download(payload)) diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index a0d03ac..92eb8c4 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -277,13 +277,60 @@ class AgentTests(unittest.TestCase): 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")): + def test_all_downloads_finish_before_any_file_is_hashed_and_verified_from_disk(self): + contents = {"zone/first.ff": b"game-file" * 200000, "zone/second.ff": b"second-file"} + manifest = self.manifest_files(contents) + self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")]) + original_hash, original_open = module.hashlib.sha256, Path.open + reads = [] + + def hashing(*args, **kwargs): + progress = self.agent.library_progress.get("downloaded") + if progress: + self.assertEqual(progress["status"], "verifying") + self.assertEqual(progress["bytesDownloaded"], manifest["sizeBytes"]) + return original_hash(*args, **kwargs) + + def opening(path, mode="r", *args, **kwargs): + if mode == "rb" and path.name.endswith(".ff"): + self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) + reads.append(path.name) + return original_open(path, mode, *args, **kwargs) + + with patch.object(module.hashlib, "sha256", side_effect=hashing), patch.object(Path, "open", opening): result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(result["state"], "ready") + self.assertCountEqual(reads, ["first.ff", "second.ff"]) + cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") + self.assertEqual(cache["bytesVerified"], manifest["sizeBytes"]) + self.assertEqual(cache["transfer"]["phase"], "verifying") + + def test_phase_boundaries_are_reported_even_between_heartbeats_and_status_errors_are_nonfatal(self): + content = b"game-file" + manifest = self.manifest(content) + reports = [] + + def fetch(method, path, data=None, extra_headers=None, timeout=60): + if path == "/status": + self.assertEqual(timeout, 5) + reports.append(json.loads(data)) + raise AgentError("Controller temporarily unavailable") + self.assertEqual(timeout, 60) + return Response(json.dumps(manifest).encode() if path == "/manifest" else content) + + self.agent.artifact_open = fetch + result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "statusUrl": "/status"}) + self.assertEqual(result["state"], "ready") + self.assertEqual([item["status"] for item in reports], ["downloading", "verifying", "ready"]) + self.assertEqual([item["bytesVerified"] for item in reports], [0, 0, len(content)]) + self.assertEqual(reports[1]["bytesDownloaded"], len(content)) + + def test_status_socket_timeout_is_short_without_reducing_download_timeout(self): + with patch.object(module.urllib.request, "build_opener") as build_opener: + self.agent.artifact_open("GET", "/file") + self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 60) + self.agent.artifact_open("POST", "/status", b"{}", timeout=5) + self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 5) def test_ignored_range_restarts_file_and_checksum(self): content = b"complete-game-file" @@ -313,6 +360,80 @@ class AgentTests(unittest.TestCase): with self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) + self.assertFalse((staging / "test.ff").exists()) + self.assertEqual(self.agent.library_progress["downloaded"]["bytesDownloaded"], 0) + requests = [] + + def fetch(method, path, data=None, extra_headers=None): + if path != "/manifest": + requests.append(extra_headers) + return Response(json.dumps(manifest).encode() if path == "/manifest" else content) + + self.agent.artifact_open = fetch + self.assertEqual(self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})["state"], "ready") + self.assertEqual(requests, [{}]) + + def test_all_corrupt_complete_files_are_removed_so_retry_downloads_them_again(self): + contents = {"zone/first.ff": b"first", "zone/second.ff": b"second", "zone/good.ff": b"good"} + manifest = self.manifest_files(contents) + staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" + staging.mkdir(parents=True) + for name, content in contents.items(): + (staging / Path(name).name).write_bytes(content if name.endswith("good.ff") else b"x" * len(content)) + requests = [] + + def fetch(method, path, data=None, extra_headers=None): + requests.append(path) + return Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")]) + + self.agent.artifact_open = fetch + payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} + with self.assertRaisesRegex(AgentError, "integrity"): + self.agent.library_download(payload) + self.assertEqual(requests, ["/manifest"]) + self.assertEqual([path.name for path in staging.iterdir()], ["good.ff"]) + self.assertEqual(self.agent.library_download(payload)["state"], "ready") + self.assertCountEqual(requests, ["/manifest", "/manifest", "/files/zone/first.ff", "/files/zone/second.ff"]) + + def test_verification_keeps_download_phase_complete_until_failure_reports_retained_bytes(self): + contents = {"zone/first.ff": b"first", "zone/second.ff": b"second"} + for mode in ("staged-corrupt", "existing-missing", "existing-short"): + with self.subTest(mode=mode): + manifest = {**self.manifest_files(contents), "id": mode} + directory = self.agent.library_dir("t6", mode) + if mode.startswith("existing"): + (directory / "content/zone").mkdir(parents=True) + (directory / "content/zone/second.ff").write_bytes(contents["zone/second.ff"]) + if mode == "existing-short": + (directory / "content/zone/first.ff").write_bytes(b"f") + module.atomic_json(directory / "manifest.json", {"libraryId": mode, "game": "t6", "size": manifest["sizeBytes"], "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]}) + + def fetch(method, path, **kwargs): + if path == "/manifest": + return Response(json.dumps(manifest).encode()) + content = contents[path.removeprefix("/files/")] + return Response(b"x" * len(content) if path.endswith("first.ff") else content) + + self.agent.artifact_open = fetch + original_open = Path.open + snapshots = [] + + def opening(path, open_mode="r", *args, **kwargs): + open_mode = kwargs.pop("mode", open_mode) + if open_mode == "rb" and path.name == "second.ff": + cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode) + snapshots.append(cache) + self.assertEqual(cache["status"], "verifying") + self.assertEqual(cache["bytesDownloaded"], cache["bytesTotal"]) + self.assertEqual(cache["transfer"]["phase"], "verifying") + return original_open(path, open_mode, *args, **kwargs) + + with patch.object(Path, "open", opening), self.assertRaisesRegex(AgentError, "integrity"): + self.agent.library_download({"libraryVersionId": mode, "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.assertEqual(len(snapshots), 1) + failed = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode) + self.assertEqual(failed["status"], "failed") + self.assertEqual(failed["bytesDownloaded"], len(contents["zone/second.ff"]) + (1 if mode == "existing-short" else 0)) 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"} @@ -369,7 +490,13 @@ class AgentTests(unittest.TestCase): 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"]}) + self.assertEqual(cache["status"], "failed") + self.assertEqual(cache["bytesDownloaded"], 8) + self.assertEqual(cache["bytesVerified"], 0) + self.assertEqual(cache["bytesTotal"], len(content)) + self.assertEqual(cache["manifestSha256"], manifest["manifestSha256"]) + self.assertEqual(cache["transfer"]["phase"], "downloading") + self.assertIsNone(cache["transfer"]["etaSeconds"]) def fetch(method, path, data=None, extra_headers=None): if path == "/manifest": @@ -405,6 +532,139 @@ class AgentTests(unittest.TestCase): self.assertEqual(caches[0]["status"], "downloading") self.assertEqual(caches[0]["bytesDownloaded"], 3) + def test_interrupted_verification_restarts_from_disk_without_redownloading_and_keeps_elapsed(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) + + def interrupted(files, targets, transfer, repair=False): + transfer.advance("bytesVerified", 5, 5) + transfer.persist(force=True) + raise KeyboardInterrupt("Simulated process termination") + + with patch.object(self.agent, "verify_downloaded_files", side_effect=interrupted), self.assertRaises(KeyboardInterrupt): + self.agent.library_download(payload) + previous = self.agent.library_transfers["downloaded"].snapshot() + self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + interrupted_cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") + self.assertEqual(interrupted_cache["status"], "missing") + self.assertEqual(interrupted_cache["bytesDownloaded"], len(content)) + self.assertEqual(interrupted_cache["bytesVerified"], 5) + requests = [] + + def fetch(method, path, **kwargs): + requests.append(path) + return Response(json.dumps(manifest).encode()) + + original_verify = self.agent.verify_downloaded_files + + def verifying(files, targets, transfer, repair=False): + self.assertEqual(transfer.snapshot()["bytesVerified"], 0) + self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) + return original_verify(files, targets, transfer, repair) + + self.agent.artifact_open = fetch + with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying): + self.assertEqual(self.agent.library_download(payload)["state"], "ready") + self.assertEqual(requests, ["/manifest"]) + cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") + self.assertEqual(cache["bytesVerified"], len(content)) + self.assertEqual(cache["transfer"]["startedAt"], previous["transfer"]["startedAt"]) + self.assertGreaterEqual(cache["transfer"]["elapsedSeconds"], interrupted_cache["transfer"]["elapsedSeconds"]) + self.assertGreaterEqual(cache["transfer"]["phaseElapsedSeconds"], interrupted_cache["transfer"]["phaseElapsedSeconds"]) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + self.assertEqual(next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"), cache) + + def test_crash_before_atomic_promotion_reverifies_read_only_staging_without_network_transfers(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) + original_replace = module.os.replace + + def replace(source, destination): + if Path(destination) == self.agent.library_dir("t6", "downloaded"): + raise KeyboardInterrupt("Simulated process termination before promotion") + return original_replace(source, destination) + + with patch.object(module.os, "replace", side_effect=replace), self.assertRaises(KeyboardInterrupt): + self.agent.library_download(payload) + staging = self.agent.root / "libraries/t6/.downloaded.partial/content" + self.assertEqual(staging.stat().st_mode & 0o777, 0o555) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + requests = [] + + def fetch(method, path, **kwargs): + requests.append(path) + return Response(json.dumps(manifest).encode()) + + self.agent.artifact_open = fetch + self.assertEqual(self.agent.library_download(payload)["state"], "ready") + self.assertEqual(requests, ["/manifest"]) + self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) + + def test_existing_cache_reverification_is_visible_and_failure_revokes_readiness_after_restart(self): + saved = json.loads((self.library / "manifest.json").read_text()) + manifest = {**saved, "id": self.library_id} + self.agent.artifact_open = lambda *args, **kwargs: Response(json.dumps(manifest).encode()) + original_verify = self.agent.verify_downloaded_files + + def verifying(files, targets, transfer, repair=False): + inventory = self.agent.inventory() + cache = next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id) + self.assertEqual(cache["status"], "verifying") + self.assertEqual(cache["bytesVerified"], 0) + self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"]) + return original_verify(files, targets, transfer, repair) + + (self.library / "content/game.ff").write_bytes(b"corrupt!!") + with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying), self.assertRaisesRegex(AgentError, "integrity"): + self.agent.library_download({"libraryVersionId": self.library_id, "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + inventory = self.agent.inventory() + self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"]) + self.assertEqual(next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)["status"], "failed") + + def test_elapsed_and_phase_eta_use_new_work_only_and_freeze_at_completion_or_failure(self): + for terminal_status in ("ready", "failed"): + with self.subTest(terminal_status=terminal_status): + now = [100.0] + progress = {"libraryVersionId": "downloaded", "status": "downloading", "bytesDownloaded": 80, "bytesVerified": 0, "bytesTotal": 100} + previous = {"transfer": {"phase": "downloading", "startedAt": "2026-10-04T00:00:00Z", "elapsedSeconds": 20, "phaseElapsedSeconds": 20}} + with patch.object(module.time, "monotonic", side_effect=lambda: now[0]): + transfer = module.LibraryTransfer(progress, self.agent.root / "agent/timing.json", previous) + self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) + now[0] += 0.5 + transfer.advance("bytesDownloaded", 5, 5) + self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) + now[0] += 1.5 + snapshot = transfer.snapshot()["transfer"] + self.assertEqual(snapshot["bytesPerSecond"], 2.5) + self.assertEqual(snapshot["etaSeconds"], 6) + self.assertEqual(snapshot["elapsedSeconds"], 22) + now[0] += 1 + transfer.advance("bytesDownloaded", 15, 15) + transfer.begin_verification() + self.assertEqual(transfer.snapshot()["transfer"]["phaseElapsedSeconds"], 0) + self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) + now[0] += 2 + transfer.advance("bytesVerified", 40, 40) + snapshot = transfer.snapshot()["transfer"] + self.assertEqual(snapshot["bytesPerSecond"], 20) + self.assertEqual(snapshot["etaSeconds"], 3) + self.assertEqual(snapshot["elapsedSeconds"], 25) + transfer.finish(terminal_status) + final = transfer.snapshot() + now[0] += 100 + self.assertEqual(transfer.snapshot(), final) + self.assertEqual(final["transfer"]["etaSeconds"], 0 if terminal_status == "ready" else None) + def test_partial_inventory_rejects_changed_manifest_and_symlinked_content(self): content = b"complete-game-file" manifest = self.manifest(content) diff --git a/docs/native-hosts.md b/docs/native-hosts.md index 49ef32b..f5d06a0 100644 --- a/docs/native-hosts.md +++ b/docs/native-hosts.md @@ -282,11 +282,23 @@ 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. Verified uploads are immutable; changing files means uploading another folder -for the same game. 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 +for the same game. Hosts first download every file with up to four concurrent +transfers and HTTP Range resumption. After all transfers finish, a separate +verification pass reads every file from disk and checks its complete SHA-256, +including any resumed prefix. Only then is the complete cache promoted atomically. +Corrupt staged files are removed so the next attempt downloads them again. +Separating the phases removes checksum work from the transfer loop and exposes +disk-verification time separately; it does not guarantee a shorter total duration. + +The Game Files and host setup pages show **Download** and **Verify** as separate +steps, each with its own progress. Elapsed time records active work across +resumptions and stops when work finishes or fails. Remaining time and speed use +the current step's measured progress; a download estimate does not include the +verification pass that follows. Estimates appear after enough progress is +measured and are hidden for stale or offline hosts. Existing agents without +timing reports still show byte progress. Agent 1.2.2 provides phase timing; +automatic updates install it between host operations. +Phase transitions are reported promptly and periodic heartbeats carry progress; 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. Previous immutable uploads remain usable diff --git a/src/lib/game-files.ts b/src/lib/game-files.ts index 88e22fa..314dbeb 100644 --- a/src/lib/game-files.ts +++ b/src/lib/game-files.ts @@ -7,10 +7,16 @@ export type GameLibraryState = 'importing' | 'ready' | 'failed'; export type LibraryCacheState = 'missing' | 'downloading' | 'verifying' | 'ready' | 'failed' | 'outdated'; export interface GameFileEntry { path: string; size: number; sha256: string } +export interface LibraryTransferProgress { + phase: 'downloading' | 'verifying'; startedAt: string; + elapsedSeconds: number; phaseElapsedSeconds: number; + bytesPerSecond: number | null; etaSeconds: number | null; +} export interface HostLibraryCache { hostId: string; libraryVersionId: string; status: LibraryCacheState; bytesDownloaded: number; bytesTotal: number; manifestSha256: string | null; error: string | null; updatedAt: string; operationId: string | null; + bytesVerified?: number; transfer?: LibraryTransferProgress; } export interface GameLibraryVersion { id: string; libraryId: string; game: GameLibraryGame; version: string; diff --git a/src/lib/hosts.ts b/src/lib/hosts.ts index 8da339e..4c607a6 100644 --- a/src/lib/hosts.ts +++ b/src/lib/hosts.ts @@ -1,3 +1,5 @@ +import type { HostLibraryCache } from './game-files.js'; + export type HostMode = 'active' | 'draining' | 'disabled'; export type HostReadiness = 'preparing' | 'needs-runtime' | 'needs-game-files' | 'ready' | 'degraded' | 'draining' | 'disabled' | 'deleting'; export type HostOperationState = 'pending' | 'accepted' | 'running' | 'completed' | 'failed' | 'unknown'; @@ -86,6 +88,7 @@ export interface HostPreparationItem { kind: 'runtime' | 'library'; label: string; state: 'waiting' | 'preparing' | 'ready' | 'blocked'; detail: string; operationId?: string; bytesDownloaded?: number; bytesTotal?: number; + cache?: HostLibraryCache; } export interface HostPreparation { state: HostPreparationItem['state']; items: HostPreparationItem[]; diff --git a/src/lib/server/game-files/index.ts b/src/lib/server/game-files/index.ts index 70792e2..e3a7cdb 100644 --- a/src/lib/server/game-files/index.ts +++ b/src/lib/server/game-files/index.ts @@ -8,14 +8,16 @@ import { ensureGameFileSchema } from './schema.js'; import { manifestChecksum, receiveGameFile, validateGamePath, validateManifest, verifyGameFile } from './storage.js'; import { MAX_UPLOAD_CHUNK_BYTES, validateUploadPlan, getUploadFileProgress, receiveUploadChunk, hashUploadedFiles, withUploadLock } from './uploads.js'; import { GAME_FILE_GAMES } from '../../game-files.js'; -import type { GameFileEntry, GameLibrary, GameLibraryGame, GameLibraryVersion, HostLibraryCache, LibraryCacheState } from '../../game-files.js'; +import type { GameFileEntry, GameLibrary, GameLibraryGame, GameLibraryVersion, HostLibraryCache, LibraryCacheState, LibraryTransferProgress } from '../../game-files.js'; function database() { const handle = db(); ensureGameFileSchema(handle); return handle; } export const gameFilesRoot = () => resolve(process.env.GAME_FILES_ROOT ?? join(dirname(config.dbPath), 'game-files')); const stagingRoot = (id: string) => join(gameFilesRoot(), '.imports', id); type VersionRow = { id: string; library_id: string; game: GameLibraryGame; version: string; description: string; dlc: string; state: GameLibraryVersion['state']; size_bytes: number; manifest: string; manifest_sha256: string | null; created_at: string; published_at: string | null; storage_location: string; source_host_id: string; source_volume: string; error: string | null; operation_id: string | null }; -type CacheRow = { host_id: string; version_id: string; status: LibraryCacheState; bytes_downloaded: number; bytes_total: number; manifest_sha256: string | null; error: string | null; updated_at: string; operation_id: string | null }; -const cacheFromRow = (row: CacheRow): HostLibraryCache => ({ hostId: row.host_id, libraryVersionId: row.version_id, status: row.status, bytesDownloaded: row.bytes_downloaded, bytesTotal: row.bytes_total, manifestSha256: row.manifest_sha256, error: row.error, updatedAt: row.updated_at, operationId: row.operation_id }); +type CacheRow = { host_id: string; version_id: string; status: LibraryCacheState; bytes_downloaded: number; bytes_total: number; manifest_sha256: string | null; error: string | null; updated_at: string; operation_id: string | null; bytes_verified: number | null; transfer_progress: string | null }; +const cacheFromRow = (row: CacheRow): HostLibraryCache => ({ hostId: row.host_id, libraryVersionId: row.version_id, status: row.status, bytesDownloaded: row.bytes_downloaded, bytesTotal: row.bytes_total, manifestSha256: row.manifest_sha256, error: row.error, updatedAt: row.updated_at, operationId: row.operation_id, + ...(row.bytes_verified == null ? {} : { bytesVerified: row.bytes_verified }), + ...(row.transfer_progress ? { transfer: JSON.parse(row.transfer_progress) as LibraryTransferProgress } : {}) }); function versionFromRow(row: VersionRow): GameLibraryVersion { const files = JSON.parse(row.manifest) as GameFileEntry[]; return { id: row.id, libraryId: row.library_id, game: row.game, version: row.version, @@ -282,6 +284,28 @@ export function requestImportPublication(hostId: string, id: string) { return importPublicationStatus(hostId, id); } +function transferProgress(value: unknown, status: LibraryCacheState): LibraryTransferProgress | null { + if (value === null || value === undefined) return null; + if (typeof value !== 'object' || Array.isArray(value)) throw new Error('Invalid transfer progress.'); + const progress = value as Record; + const nonnegative = (amount: unknown): amount is number => typeof amount === 'number' && Number.isFinite(amount) && amount >= 0 && amount <= Number.MAX_SAFE_INTEGER; + if (!['downloading', 'verifying'].includes(progress.phase as string) + || typeof progress.startedAt !== 'string' || progress.startedAt.length > 80 + || !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})$/.test(progress.startedAt) + || !Number.isFinite(Date.parse(progress.startedAt)) + || !nonnegative(progress.elapsedSeconds) || !nonnegative(progress.phaseElapsedSeconds) + || progress.phaseElapsedSeconds > progress.elapsedSeconds + || (progress.bytesPerSecond != null && !nonnegative(progress.bytesPerSecond)) + || (progress.etaSeconds != null && !nonnegative(progress.etaSeconds)) + || (['downloading', 'verifying'].includes(status) && progress.phase !== status) + || (['ready', 'outdated'].includes(status) && progress.phase !== 'verifying')) { + throw new Error('Invalid transfer progress.'); + } + return { phase: progress.phase as LibraryTransferProgress['phase'], startedAt: progress.startedAt, + elapsedSeconds: progress.elapsedSeconds, phaseElapsedSeconds: progress.phaseElapsedSeconds, + bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null }; +} + export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void { if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.'); const handle = database(); @@ -303,15 +327,27 @@ export function updateHostLibraryCaches(hostId: string, input: unknown, options: const bytes = raw.bytesDownloaded ?? (['ready', 'outdated'].includes(status) ? version.sizeBytes : 0); if (!Number.isSafeInteger(bytes) || bytes < 0 || bytes > version.sizeBytes) throw new Error('Invalid library download progress.'); if (['ready', 'outdated'].includes(status) && bytes !== version.sizeBytes) { status = 'failed'; error = 'Host reported an incomplete library as verified.'; } - handle.prepare(`INSERT INTO host_library_caches(host_id,version_id,status,bytes_downloaded,bytes_total,manifest_sha256,error,updated_at) - VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(host_id,version_id) DO UPDATE SET status=excluded.status, - bytes_downloaded=excluded.bytes_downloaded,bytes_total=excluded.bytes_total,manifest_sha256=excluded.manifest_sha256,error=excluded.error,updated_at=excluded.updated_at`) - .run(hostId, id, status, bytes, version.sizeBytes, checksum, error, now); + const prior = getHostLibraryCache(hostId, id); + const terminal = ['ready', 'outdated', 'failed'].includes(status); + // Older agents do not report timing. Keep a completed measurement on + // later inventory heartbeats, but never carry an ETA into a new phase. + let transfer = raw.transfer === undefined && terminal ? prior?.transfer ?? null : transferProgress(raw.transfer, status); + let verified = raw.bytesVerified ?? (['ready', 'outdated'].includes(status) ? version.sizeBytes : terminal ? prior?.bytesVerified ?? null : null); + if (verified !== null && (!Number.isSafeInteger(verified) || verified < 0 || verified > version.sizeBytes)) throw new Error('Invalid library verification progress.'); + if (transfer && status === 'verifying' && bytes !== version.sizeBytes) throw new Error('Download must finish before verification starts.'); + if (['ready', 'outdated'].includes(status) && verified !== version.sizeBytes) throw new Error('Host reported incomplete verification as ready.'); + if (status === 'missing') { transfer = null; verified = null; } + if (terminal && transfer) transfer = { ...transfer, bytesPerSecond: null, etaSeconds: null }; + handle.prepare(`INSERT INTO host_library_caches(host_id,version_id,status,bytes_downloaded,bytes_total,manifest_sha256,error,updated_at,bytes_verified,transfer_progress) + VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(host_id,version_id) DO UPDATE SET status=excluded.status, + bytes_downloaded=excluded.bytes_downloaded,bytes_total=excluded.bytes_total,manifest_sha256=excluded.manifest_sha256,error=excluded.error,updated_at=excluded.updated_at, + bytes_verified=excluded.bytes_verified,transfer_progress=excluded.transfer_progress`) + .run(hostId, id, status, bytes, version.sizeBytes, checksum, error, now, verified, transfer ? JSON.stringify(transfer) : null); } if (options.snapshot) { const cached = handle.prepare('SELECT version_id FROM host_library_caches WHERE host_id=?').all(hostId) as { version_id: string }[]; for (const cache of cached) if (!reported.has(cache.version_id)) { - handle.prepare("UPDATE host_library_caches SET status='missing',bytes_downloaded=0,manifest_sha256=NULL,error=NULL,updated_at=? WHERE host_id=? AND version_id=?") + handle.prepare("UPDATE host_library_caches SET status='missing',bytes_downloaded=0,manifest_sha256=NULL,error=NULL,bytes_verified=NULL,transfer_progress=NULL,updated_at=? WHERE host_id=? AND version_id=?") .run(now, hostId, cache.version_id); } } @@ -336,7 +372,7 @@ export function queueLibraryDownload(hostId: string, versionId: string, options: manifestSha256: version.manifestSha256, sizeBytes: version.sizeBytes, ...(options.automatic ? { automatic: true } : {}) }); - updateHostLibraryCaches(hostId, [{ libraryVersionId: versionId, status: 'downloading', bytesDownloaded: retainedBytes, manifestSha256: prior?.manifestSha256 }]); + updateHostLibraryCaches(hostId, [{ libraryVersionId: versionId, status: 'downloading', bytesDownloaded: retainedBytes, manifestSha256: prior?.manifestSha256, bytesVerified: 0, transfer: null }]); 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/game-files/schema.ts b/src/lib/server/game-files/schema.ts index ae62c16..a412ef0 100644 --- a/src/lib/server/game-files/schema.ts +++ b/src/lib/server/game-files/schema.ts @@ -40,5 +40,8 @@ export function ensureGameFileSchema(handle: Database.Database): void { CREATE TRIGGER IF NOT EXISTS game_file_versions_immutable BEFORE UPDATE ON game_file_versions WHEN OLD.state = 'ready' BEGIN SELECT RAISE(ABORT, 'Published game-file versions are immutable'); END; `); + const cacheColumns = handle.prepare('PRAGMA table_info(host_library_caches)').all() as { name: string }[]; + if (!cacheColumns.some(column => column.name === 'bytes_verified')) handle.exec('ALTER TABLE host_library_caches ADD COLUMN bytes_verified INTEGER'); + if (!cacheColumns.some(column => column.name === 'transfer_progress')) handle.exec('ALTER TABLE host_library_caches ADD COLUMN transfer_progress TEXT'); initialized.add(handle); } diff --git a/src/lib/server/hosts/preparation.ts b/src/lib/server/hosts/preparation.ts index 6e77442..6b8a8a2 100644 --- a/src/lib/server/hosts/preparation.ts +++ b/src/lib/server/hosts/preparation.ts @@ -56,7 +56,7 @@ function targets(host: Host): Target[] { operation: latestOperation(host.id, 'library.download', 'libraryVersionId', version.id), item: { kind: 'library', label: version.name, state: ready ? 'ready' : 'waiting', detail: ready ? 'The latest game files are downloaded and verified.' : 'The latest game files will be downloaded and verified automatically.', - bytesDownloaded: cache?.bytesDownloaded ?? 0, bytesTotal: version.sizeBytes } }); + bytesDownloaded: cache?.bytesDownloaded ?? 0, bytesTotal: version.sizeBytes, ...(cache ? { cache } : {}) } }); } const blocker = hostBlocker(host); for (const target of result) { @@ -67,7 +67,8 @@ function targets(host: Host): Target[] { 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.'; + : target.image ? 'Downloading the runtime automatically.' + : target.item.cache?.status === 'verifying' ? 'Download complete. Verifying all game files.' : 'Downloading game files. Verification starts after all files arrive.'; } else if (operation && Date.parse(host.lastHeartbeat!) <= Date.parse(operation.updated_at)) { target.item.state = 'preparing'; target.item.detail = 'Waiting for a fresh host inventory after preparation.'; } else if (retryAt(host.id, target) > Date.now()) { @@ -110,7 +111,7 @@ function supersedeUndeliveredDownloads(hostId: string): void { // automatic work can be replaced without consulting the agent journal. db().prepare("UPDATE ec_host_operations SET state='failed',error=?,updated_at=? WHERE id=?").run(supersededDownload, now, operation.id); db().prepare("INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,'failed',?,?)").run(operation.id, now, supersededDownload); - db().prepare("UPDATE host_library_caches SET status='missing',error=NULL,updated_at=? WHERE host_id=? AND version_id=? AND operation_id=? AND status NOT IN ('ready','outdated')") + db().prepare("UPDATE host_library_caches SET status='missing',error=NULL,bytes_verified=NULL,transfer_progress=NULL,updated_at=? WHERE host_id=? AND version_id=? AND operation_id=? AND status NOT IN ('ready','outdated')") .run(now, hostId, operation.versionId, operation.id); } } diff --git a/src/lib/ui/components/GameFileTransferProgress.svelte b/src/lib/ui/components/GameFileTransferProgress.svelte new file mode 100644 index 0000000..cc95557 --- /dev/null +++ b/src/lib/ui/components/GameFileTransferProgress.svelte @@ -0,0 +1,52 @@ + + +
+
    +
  1. 1. {downloaded ? 'Downloaded' : downloading ? 'Downloading' : 'Download'}
    2. +
  2. 2. {progress.complete ? 'Verified' : verifying ? 'Verifying' : 'Verify'}
    3. +
+ {#if progress.phase && !progress.complete} + + {#if progress.phaseBytes !== null}{formatGameBytes(progress.phaseBytes)} / {formatGameBytes(progress.total)} {progress.phase === 'verifying' ? 'checked' : 'downloaded'}{:else}Checking every file…{/if} + {/if} + {#if progress.elapsedSeconds !== null} +
+ Elapsed {formatDuration(progress.elapsedSeconds * 1000)} + {#if progress.bytesPerSecond !== null}{formatGameBytes(progress.bytesPerSecond)}/s{/if} + {#if progress.etaSeconds !== null}About {formatDuration(Math.ceil(progress.etaSeconds) * 1000)} {progress.phase === 'verifying' ? 'verification' : 'download'} remaining + {:else if progress.live && progress.phaseBytes !== progress.total}Estimating {progress.phase === 'verifying' ? 'verification' : 'download'} time remaining…{/if} +
+ {/if} + {#if progress.active && !connected}Host offline · waiting to resume updates + {:else if progress.stale}Waiting for a fresh host report + {:else if downloading}All files download before verification starts.{/if} +
+ + diff --git a/src/lib/ui/components/HostPreparation.svelte b/src/lib/ui/components/HostPreparation.svelte index e45ab50..1a056c7 100644 --- a/src/lib/ui/components/HostPreparation.svelte +++ b/src/lib/ui/components/HostPreparation.svelte @@ -1,6 +1,7 @@ @@ -13,7 +14,9 @@
  • {item.label}{item.state === 'blocked' ? 'Needs attention' : item.state}

    {item.detail}

    - {#if item.bytesTotal && item.bytesDownloaded !== undefined && item.state === 'preparing'} + {#if item.cache && item.cache.status !== 'missing'} +
    + {:else if item.bytesTotal && item.bytesDownloaded !== undefined && item.state === 'preparing'} {formatGameBytes(item.bytesDownloaded)} / {formatGameBytes(item.bytesTotal)} {/if} @@ -28,4 +31,5 @@ .preparation { display:grid; gap:12px; } p { margin:0; color:var(--text-secondary); font-size:var(--text-sm); line-height:1.6; } .status { color:var(--text-primary); } ul { list-style:none; padding:0; margin:0; display:grid; gap:16px; } li { min-width:0; } .row { display:flex; justify-content:space-between; gap:12px; font-size:var(--text-sm); } strong { overflow-wrap:anywhere; } span { text-transform:capitalize; color:var(--text-secondary); } li p { margin-top:4px; } .blocked { color:var(--state-danger); } progress { width:100%; margin-top:8px; accent-color:var(--accent-primary); } small { display:block; color:var(--text-secondary); font-size:var(--text-xs); } a { color:var(--accent-primary-hover); } + .transfer { margin-top:8px; } diff --git a/src/lib/ui/game-file-transfer.ts b/src/lib/ui/game-file-transfer.ts new file mode 100644 index 0000000..02622e9 --- /dev/null +++ b/src/lib/ui/game-file-transfer.ts @@ -0,0 +1,33 @@ +import type { HostLibraryCache } from '../game-files.js'; + +export const TRANSFER_REPORT_FRESH_MS = 30_000; + +const nonnegative = (value: number | null | undefined): number | null => + typeof value === 'number' && Number.isFinite(value) && value >= 0 ? value : null; + +/** Keep byte progress factual; only the elapsed clock advances between host reports. */ +export function gameFileTransferProgress(cache: HostLibraryCache, connected: boolean, now: number) { + const active = cache.status === 'downloading' || cache.status === 'verifying'; + const complete = cache.status === 'ready' || cache.status === 'outdated'; + const phase = active ? cache.status as 'downloading' | 'verifying' : cache.transfer?.phase ?? null; + const total = nonnegative(cache.bytesTotal) ?? 0; + const downloaded = Math.min(total, nonnegative(cache.bytesDownloaded) ?? 0); + const verified = complete ? total : nonnegative(cache.bytesVerified); + const phaseBytes = phase === 'verifying' ? verified === null ? null : Math.min(total, verified) : downloaded; + const transfer = cache.transfer; + const samePhase = transfer?.phase === phase; + const reportedAt = Date.parse(cache.updatedAt); + const reportAge = Number.isFinite(reportedAt) && Number.isFinite(now) ? Math.max(0, now - reportedAt) : null; + const stale = active && !!transfer && (reportAge === null || reportAge > TRANSFER_REPORT_FRESH_MS); + const live = active && connected && samePhase && !stale && reportAge !== null; + const elapsed = nonnegative(transfer?.elapsedSeconds); + // Stop extrapolating when reports stop. A disconnected or finished host uses its saved duration. + const elapsedSeconds = elapsed === null ? null : elapsed + (active && connected && samePhase && reportAge !== null + ? Math.min(reportAge, TRANSFER_REPORT_FRESH_MS) / 1000 : 0); + const rate = nonnegative(transfer?.bytesPerSecond); + const estimate = nonnegative(transfer?.etaSeconds); + const hasProgress = phaseBytes !== null && phaseBytes > 0 && phaseBytes < total; + const bytesPerSecond = live && hasProgress && rate !== null && rate > 0 ? rate : null; + const etaSeconds = bytesPerSecond !== null && estimate !== null && estimate > 0 ? estimate : null; + return { active, complete, phase, total, downloaded, verified, phaseBytes, elapsedSeconds, bytesPerSecond, etaSeconds, stale, live }; +} diff --git a/src/routes/game-files/+page.svelte b/src/routes/game-files/+page.svelte index acbe2ab..92163bc 100644 --- a/src/routes/game-files/+page.svelte +++ b/src/routes/game-files/+page.svelte @@ -10,6 +10,7 @@ import SectionHeader from '$ui/components/primitives/SectionHeader.svelte'; import Button from '$ui/components/primitives/Button.svelte'; import Badge from '$ui/components/primitives/Badge.svelte'; + import GameFileTransferProgress from '$ui/components/GameFileTransferProgress.svelte'; type OverviewLibrary = Omit & { versions: Omit[] }; interface UploadFile { path: string; size: number; lastModified: number; receivedBytes: number; complete: boolean } @@ -292,7 +293,7 @@ {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' ? ' · previous upload' : ''}{:else}—{/if} + {#if cache.status !== 'missing'}{:else}—{/if} {cache.error ?? (awaitingAutomaticDownload ? automaticDownloadDetail(cacheHost) : '')}{#if awaitingAutomaticDownload || cache.status === 'failed'} View host setup{/if} {:else}No host copies yet.{/each} @@ -315,8 +316,8 @@ .path-list { padding-left:20px; font-family:var(--font-mono); font-size:var(--text-xs); overflow-wrap:anywhere; } .current-file { font-family:var(--font-mono); font-size:var(--text-xs); overflow-wrap:anywhere; } progress { width:100%; accent-color:var(--accent-primary); } h2,h3,p { margin:0; } h2 { font-size:var(--text-md); } h3 { font-size:var(--text-sm); } - p,td,th,details { font-size:var(--text-sm); line-height:1.6; } .muted,small { color:var(--text-secondary); } - small { display:block; } a { color:var(--accent-primary-hover); } summary { cursor:pointer; } + p,td,th,details { font-size:var(--text-sm); line-height:1.6; } .muted { color:var(--text-secondary); } + a { color:var(--accent-primary-hover); } summary { cursor:pointer; } dl { display:grid; grid-template-columns:auto 1fr; gap:6px 16px; } dd { margin:0; overflow-wrap:anywhere; } .table-wrap { overflow-x:auto; } table { width:100%; border-collapse:collapse; text-align:left; } th,td { padding:10px; border-bottom:1px solid var(--border-subtle); } @media(max-width:640px) { .card,.folder-summary,.upload-progress { padding:16px; } dl { grid-template-columns:1fr; } .heading>div { min-width:0; } } diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 0ae344f..22f605e 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -30,7 +30,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.2.1" +VERSION = "1.2.2" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 LIBRARY_DOWNLOAD_WORKERS = 4 @@ -587,6 +587,80 @@ class Uninstaller: time.sleep(15) +class LibraryTransfer: + """Durable phase progress; throughput counts only work in this attempt.""" + + def __init__(self, progress, marker, previous=None): + self.progress = progress + self.marker = marker + self.lock = threading.RLock() + prior = (previous or {}).get("transfer") or {} + self.prior_phase = prior.get("phase") + self.prior_phase_elapsed = prior.get("phaseElapsedSeconds", 0) + self.started_at = prior.get("startedAt") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + self.elapsed = prior.get("elapsedSeconds", 0) + self.phase_elapsed = prior.get("phaseElapsedSeconds", 0) if prior.get("phase") == progress["status"] else 0 + self.started = self.phase_started = time.monotonic() + self.phase = progress["status"] + self.work = 0 + self.finished = False + self.failed_bytes_downloaded = None + self.persisted_at = self.started + self.persist(force=True) + + def snapshot(self): + with self.lock: + if not self.finished: + now = time.monotonic() + sample_seconds = max(0, now - self.phase_started) + rate = self.work / sample_seconds if sample_seconds >= 1 and self.work else None + done = self.progress["bytesDownloaded"] if self.phase == "downloading" else self.progress["bytesVerified"] + self.progress["transfer"] = { + "phase": self.phase, "startedAt": self.started_at, + "elapsedSeconds": self.elapsed + max(0, now - self.started), + "phaseElapsedSeconds": self.phase_elapsed + sample_seconds, + "bytesPerSecond": rate, + "etaSeconds": max(0, self.progress["bytesTotal"] - done) / rate if rate else None, + } + return {**self.progress, "transfer": self.progress["transfer"].copy()} + + def persist(self, force=False): + with self.lock: + now = time.monotonic() + self.snapshot() + if force or now - self.persisted_at >= 5: + atomic_json(self.marker, self.progress) + self.persisted_at = now + + def advance(self, key, count, work): + with self.lock: + self.progress[key] += count + self.work += work + self.persist() + + def begin_verification(self): + with self.lock: + self.progress.update(status="verifying", bytesVerified=0) + if self.phase != "verifying": + self.phase_elapsed = self.prior_phase_elapsed if self.prior_phase == "verifying" else 0 + self.phase = "verifying" + self.phase_started = time.monotonic() + self.work = 0 + self.persist(force=True) + + def finish(self, status, error=None): + with self.lock: + self.snapshot() + self.finished = True + self.progress["status"] = status + if status == "failed" and self.failed_bytes_downloaded is not None: + self.progress["bytesDownloaded"] = self.failed_bytes_downloaded + self.progress["transfer"]["etaSeconds"] = 0 if status == "ready" else None + if error: + self.progress["error"] = error + self.persist(force=True) + + class Agent: def __init__(self, config, docker=None): self.config = config @@ -599,6 +673,7 @@ class Agent: self.stopping = threading.Event() self.current_operation = None self.library_progress = {} + self.library_transfers = {} self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None @@ -789,6 +864,8 @@ class Agent: except (AgentError, ValueError): pass libraries, caches = [], [] + progress = {key: value.copy() for key, value in list(self.library_progress.items())} + progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())}) 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. @@ -796,12 +873,22 @@ class Agent: continue try: value = json.loads(marker.read_text()) + # Rechecking an existing cache must revoke readiness until its + # entire contents have passed verification again. + if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists(): + saved_progress = json.loads((marker.parent / "transfer.json").read_text()) + if saved_progress.get("status") != "ready": + progress[value["libraryId"]] = {**saved_progress, "status": "failed" if saved_progress.get("status") == "failed" else "missing"} + if value["libraryId"] in progress: + continue libraries.append({"id": value["libraryId"], "state": "ready"}) - caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}) + cache = {"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesVerified": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")} + if value.get("transfer"): + cache["transfer"] = value["transfer"] + caches.append(cache) except (OSError, ValueError, KeyError): continue 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()) @@ -826,7 +913,15 @@ class Agent: 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"]} + cache = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": value["manifestSha256"]} + transfer_marker = marker.parent / "transfer.json" + if transfer_marker.exists(): + saved_progress = json.loads(transfer_marker.read_text()) + if saved_progress.get("manifestSha256") == value["manifestSha256"]: + cache.update(bytesVerified=saved_progress.get("bytesVerified", 0), transfer=saved_progress.get("transfer")) + if saved_progress.get("status") == "failed": + cache.update(status="failed", error=saved_progress.get("error")) + progress[library_id] = cache except (AgentError, OSError, ValueError, KeyError, TypeError): continue progress = list(progress.values()) @@ -1221,14 +1316,14 @@ class Agent: raise AgentError("Library transport must use HTTPS") return url - def artifact_open(self, method, path, data=None, extra_headers=None): + def artifact_open(self, method, path, data=None, extra_headers=None, timeout=60): headers = {"Authorization": "Bearer " + self.config["credential"], **(extra_headers or {})} req = urllib.request.Request(self.artifact_url(path), data=data, method=method, headers=headers) # Redirects cannot forward host credentials to another origin. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, hdrs, newurl): raise AgentError("Artifact redirects are not allowed") - return urllib.request.build_opener(NoRedirect()).open(req, timeout=60) + return urllib.request.build_opener(NoRedirect()).open(req, timeout=timeout) def download_library_file(self, item, target, files_url, update_progress, aborted): def check_active(): @@ -1241,21 +1336,11 @@ class Agent: 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) + # Saved prefixes and complete files are intentionally not read here. + # The second phase verifies every byte after all transfers have ended. 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) + return update_progress(item["path"], offset) check_active() url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/") @@ -1269,7 +1354,6 @@ class Agent: 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 @@ -1282,13 +1366,48 @@ class Agent: 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"]: + if target.stat().st_size != item["size"]: + raise AgentError("Downloaded library file size failed integrity verification") + + def verify_downloaded_files(self, files, targets, transfer, repair=False): + invalid = False + retained = transfer.progress["bytesDownloaded"] + for item in files: + target = targets[item["path"]] + checksum = hashlib.sha256() + actual_size = target.stat().st_size if target.is_file() else 0 + valid_size = target.is_file() and actual_size == item["size"] + if valid_size: + with target.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + checksum.update(chunk) + transfer.advance("bytesVerified", len(chunk), len(chunk)) + if not valid_size or checksum.hexdigest() != item["sha256"]: + invalid = True + if repair and target.exists(): + # A corrupt complete file (including a corrupt saved + # prefix) must not be resumed indefinitely on every retry. + target.unlink() + actual_size = 0 + retained -= item["size"] - min(actual_size, item["size"]) + if invalid: + # Verification describes a completed download phase. Keep that + # total stable while other files are checked, then publish retained + # bytes atomically with failure (including removed corrupt files). + transfer.failed_bytes_downloaded = retained raise AgentError("Downloaded library file failed integrity verification") + def publish_library_progress(self, payload, transfer): + if payload.get("statusUrl"): + # Phase boundaries are useful even when shorter than the heartbeat + # interval. Telemetry failure must never undo a verified download. + with contextlib.suppress(AgentError, OSError, ValueError): + with self.artifact_open("POST", payload["statusUrl"], json.dumps(transfer.snapshot()).encode(), {"Content-Type": "application/json"}, timeout=5) as response: + response.read() + def library_download(self, payload): library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId"))) with self.artifact_open("GET", payload["manifestUrl"]) as response: @@ -1305,12 +1424,12 @@ class Agent: if payload.get("manifestSha256") and payload["manifestSha256"] != manifest_sha: raise AgentError("Library manifest differs from the requested immutable version") directory = self.library_dir(manifest["game"], library_id) - if directory.exists(): - saved = self.verify_library(directory) + existing = directory.exists() + if existing: + saved = json.loads((directory / "manifest.json").read_text()) if saved.get("manifestSha256") != manifest_sha: raise AgentError("Published library changed unexpectedly") - return {"libraryId": library_id, "state": "ready", "size": saved["size"]} - staging = directory.with_name("." + library_id + ".partial") + staging = directory if existing else directory.with_name("." + library_id + ".partial") staging.mkdir(parents=True, exist_ok=True) source_marker = staging / "source.json" if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha: @@ -1318,28 +1437,39 @@ class Agent: size = sum(item["size"] for item in files) 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} + progress = {"libraryVersionId": library_id, "status": "verifying" if existing else "downloading", "bytesDownloaded": size if existing else 0, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": manifest_sha} self.library_progress[library_id] = progress + marker = staging / "transfer.json" + previous = json.loads(marker.read_text()) if marker.exists() else None + if previous and (previous.get("manifestSha256") != manifest_sha or previous.get("status") == "ready"): + previous = None + timing = LibraryTransfer(progress, marker, previous) + self.library_transfers[library_id] = timing 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 targets 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 type(item["size"]) is not int or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]): raise AgentError("Invalid library file manifest") 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 = {} + offsets = {item["path"]: min(targets[item["path"]].stat().st_size, item["size"]) if targets[item["path"]].exists() else 0 for item in files} + if not existing: + progress["bytesDownloaded"] = sum(offsets.values()) + if not existing and shutil.disk_usage(staging).free < size - progress["bytesDownloaded"]: + raise AgentError("Insufficient library storage") + timing.persist(force=True) + self.publish_library_progress(payload, timing) aborted = threading.Event() def update_progress(path, offset): with progress_lock: - progress["bytesDownloaded"] += offset - offsets.get(path, 0) + delta = offset - offsets.get(path, 0) offsets[path] = offset + timing.advance("bytesDownloaded", delta, max(0, delta)) def transfer(item): try: @@ -1350,21 +1480,44 @@ class Agent: # 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" + if not existing: + # A crash after setting read-only permissions but before the + # atomic rename leaves a valid, resumable staging directory. + content_root = staging / "content" + if content_root.exists(): + os.chmod(content_root, 0o755) + for path in content_root.rglob("*"): + if path.is_symlink(): + raise AgentError("Symlinks are not allowed in library content") + os.chmod(path, 0o755 if path.is_dir() else 0o644) + 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() + timing.begin_verification() + self.publish_library_progress(payload, timing) + self.verify_downloaded_files(files, targets, timing, repair=not existing) 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("*"): - os.chmod(path, 0o555 if path.is_dir() else 0o444) - os.chmod(staging / "content", 0o555) - os.replace(staging, directory) + if not existing: + for path in (staging / "content").rglob("*"): + if path.is_symlink(): + raise AgentError("Symlinks are not allowed in library content") + os.chmod(path, 0o555 if path.is_dir() else 0o444) + os.chmod(staging / "content", 0o555) + os.replace(staging, directory) + timing.marker = directory / "transfer.json" + timing.finish("ready") + saved["transfer"] = timing.snapshot()["transfer"] + saved["bytesVerified"] = size + atomic_json(directory / "manifest.json", saved) + self.publish_library_progress(payload, timing) self.library_progress.pop(library_id, None) + self.library_transfers.pop(library_id, None) return {"libraryId": library_id, "state": "ready", "size": size, "manifestSha256": manifest_sha} except Exception: - progress.update(status="failed", error="Library download or integrity verification failed") + timing.finish("failed", "Library download or integrity verification failed") + self.publish_library_progress(payload, timing) raise def library_import(self, payload): @@ -1561,14 +1714,9 @@ class Agent: finally: self.executing_id = None if kind == "library.download": - library_id = payload.get("libraryVersionId", payload.get("libraryId")) - for game in ("t5", "t6"): - directory = self.library_dir(game, library_id) - if directory.exists(): - saved = self.verify_library(directory) - return self.journal.update(operation["id"], "completed", {"libraryId": library_id, "state": "ready", "size": saved["size"], "manifestSha256": saved.get("manifestSha256"), "reconciled": True}) # This is a resumable immutable transfer, not a repeated - # container command: verified completed files remain untouched. + # container command. Full files skip transfer; verification + # always starts again so interrupted reads cannot prove ready. self.journal.update(operation["id"], "running") try: return self.journal.update(operation["id"], "completed", self.library_download(payload)) diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 57935aa..60a70cb 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.2.1", + "version": "1.2.2", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "680c0c2bd6d703d3b0119ff5b253f1f6b70d1eeee98fe0e6770738eab2b6d380", + "sha256": "c266a8099130051ee540cdc3b9cdc7e42144fa67e21cce76eea6cf0a426bf40d", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/game-file-transfer-progress.test.ts b/tests/unit/game-file-transfer-progress.test.ts new file mode 100644 index 0000000..1dcb08a --- /dev/null +++ b/tests/unit/game-file-transfer-progress.test.ts @@ -0,0 +1,236 @@ +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 { 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'); +let database: typeof import('../../src/lib/server/db/index.js'); +let hostId: string; +let version: GameLibraryVersion; +const content = 'a'.repeat(100); +const startedAt = '2026-10-04T12:00:00.000Z'; +const runtime = 'ghcr.io/example/plutainer:transfer-test'; + +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(new Date(startedAt)); + 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'); + database = await import('../../src/lib/server/db/index.js'); + const host = hosts.createHost({ name: 'Transfer progress host' }); + hostId = host.id; + hosts.enrollHost(hosts.createEnrollment(hostId).token, { + protocolVersion: 1, dockerRunning: true, directoriesWritable: true, + storageAvailable: 100 * 1024 ** 3, runtimeImages: [runtime] + }); + version = await publish(); +}); + +afterEach(async () => { + vi.useRealTimers(); + vi.unstubAllEnvs(); + await fixture.cleanup(); +}); + +async function publish(game: 't5' | 't6' = 't5') { + const upload = libraries.createLibraryUpload({ + game, files: [{ path: 'zone/game.ff', size: content.length, lastModified: 0 }] + }); + await libraries.uploadBrowserFile(upload.id, 'zone/game.ff', { + offset: 0, size: content.length, sha256: createHash('sha256').update(content).digest('hex') + }, new Response(content).body); + return libraries.publishBrowserUpload(upload.id); +} + +function transfer(phase: 'downloading' | 'verifying' = 'downloading') { + return { + phase, startedAt, elapsedSeconds: 10, + phaseElapsedSeconds: phase === 'downloading' ? 10 : 2, + bytesPerSecond: 5, etaSeconds: 10 + }; +} + +function report(overrides: Record = {}) { + return { + libraryVersionId: version.id, status: 'downloading', + bytesDownloaded: 50, bytesVerified: 0, + manifestSha256: version.manifestSha256, transfer: transfer(), ...overrides + }; +} + +function cache() { return libraries.getHostLibraryCache(hostId, version.id)!; } + +describe('host game-file transfer progress', () => { + it('persists both phases and exposes them in the library and host preparation views', () => { + libraries.queueLibraryDownload(hostId, version.id); + libraries.updateHostLibraryCaches(hostId, [report()]); + const downloading = cache(); + expect(downloading).toMatchObject({ + status: 'downloading', bytesDownloaded: 50, bytesVerified: 0, transfer: transfer() + }); + database.closeDb(); + expect(cache()).toEqual(downloading); + + const verification = transfer('verifying'); + libraries.updateHostLibraryCaches(hostId, [report({ + status: 'verifying', bytesDownloaded: 100, bytesVerified: 20, transfer: verification + })]); + const verifying = cache(); + expect(verifying).toMatchObject({ + status: 'verifying', bytesDownloaded: 100, bytesVerified: 20, transfer: verification + }); + const overview = libraries.libraryOverview().libraries.find(library => library.game === 't5'); + expect(overview!.versions.find(item => item.id === version.id)!.caches).toContainEqual(verifying); + expect(preparation.hostPreparationStatus(hosts.getHost(hostId)!).items).toContainEqual( + expect.objectContaining({ kind: 'library', state: 'preparing', cache: verifying }) + ); + expect(() => libraries.requireReadyLibrary(hostId, version.id)).toThrow(/finish downloading and verifying/); + }); + + it('allows an unknown transfer rate and ETA until the agent has measured progress', () => { + const initial = { + ...transfer(), elapsedSeconds: 0, phaseElapsedSeconds: 0, bytesPerSecond: null, etaSeconds: null + }; + libraries.updateHostLibraryCaches(hostId, [report({ bytesDownloaded: 0, transfer: initial })]); + expect(cache()).toMatchObject({ bytesDownloaded: 0, bytesVerified: 0, transfer: initial }); + }); + + it('retains completed timing when later ready heartbeats omit transfer metadata', () => { + const completed = { ...transfer('verifying'), elapsedSeconds: 30, phaseElapsedSeconds: 10, etaSeconds: 0 }; + const frozen = { ...completed, bytesPerSecond: null, etaSeconds: null }; + libraries.updateHostLibraryCaches(hostId, [report({ + status: 'ready', bytesDownloaded: 100, bytesVerified: 100, transfer: completed + })]); + vi.setSystemTime(Date.now() + 60_000); + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'ready', manifestSha256: version.manifestSha256 + }], { snapshot: true }); + expect(cache()).toMatchObject({ + status: 'ready', bytesDownloaded: 100, bytesVerified: 100, transfer: frozen + }); + expect(libraries.requireReadyLibrary(hostId, version.id).id).toBe(version.id); + database.closeDb(); + expect(cache().transfer).toEqual(frozen); + }); + + it('clears the previous transfer when the host no longer reports the library', () => { + libraries.updateHostLibraryCaches(hostId, [report()]); + libraries.updateHostLibraryCaches(hostId, [], { snapshot: true }); + expect(cache()).toMatchObject({ status: 'missing', bytesDownloaded: 0, manifestSha256: null }); + expect(cache().bytesVerified ?? 0).toBe(0); + expect(cache().transfer).toBeUndefined(); + }); + + it('clears old timing and verification when retrying while retaining resumable download bytes', () => { + const failed = libraries.queueLibraryDownload(hostId, version.id); + libraries.updateHostLibraryCaches(hostId, [report({ status: 'failed', error: 'Connection reset' })]); + hosts.reportHostOperation(hostId, failed.id, { state: 'failed', error: 'Connection reset' }); + const retry = libraries.queueLibraryDownload(hostId, version.id); + expect(retry.id).not.toBe(failed.id); + expect(cache()).toMatchObject({ status: 'downloading', bytesDownloaded: 50, operationId: retry.id }); + expect(cache().bytesVerified ?? 0).toBe(0); + expect(cache().transfer).toBeUndefined(); + }); + + it('continues accepting older agents without phase metadata', () => { + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'downloading', bytesDownloaded: 25, + manifestSha256: version.manifestSha256 + }]); + expect(cache()).toMatchObject({ status: 'downloading', bytesDownloaded: 25 }); + expect(cache().transfer).toBeUndefined(); + // Earlier agents verified files as they downloaded, so a verifying report + // could still contain a partial aggregate download count. + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'verifying', bytesDownloaded: 25, + manifestSha256: version.manifestSha256 + }]); + expect(cache().status).toBe('verifying'); + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'ready', manifestSha256: version.manifestSha256 + }]); + expect(cache()).toMatchObject({ status: 'ready', bytesDownloaded: 100 }); + expect(libraries.requireReadyLibrary(hostId, version.id).id).toBe(version.id); + }); + + it('does not leave an old ETA attached to a new report from an older agent', () => { + libraries.updateHostLibraryCaches(hostId, [report()]); + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'downloading', bytesDownloaded: 60, + manifestSha256: version.manifestSha256 + }]); + expect(cache()).toMatchObject({ status: 'downloading', bytesDownloaded: 60 }); + expect(cache().transfer).toBeUndefined(); + }); + + it.each([ + ['negative verified bytes', { bytesVerified: -1 }], + ['verified bytes exceed the library', { bytesVerified: 101 }], + ['fractional verified bytes', { bytesVerified: 0.5 }], + ['nonfinite verified bytes', { bytesVerified: Number.POSITIVE_INFINITY }], + ['downloaded bytes exceed the library', { bytesDownloaded: 101 }], + ['a partial download in the verification phase', { status: 'verifying', transfer: transfer('verifying') }], + ['ready before verification finishes', { status: 'ready', bytesDownloaded: 100, bytesVerified: 20, transfer: transfer('verifying') }], + ['a phase inconsistent with status', { transfer: transfer('verifying') }], + ['an unknown phase', { transfer: { ...transfer(), phase: 'uploading' } }], + ['a malformed start time', { transfer: { ...transfer(), startedAt: 'not-a-timestamp' } }], + ['a non-string start time', { transfer: { ...transfer(), startedAt: 1234 } }], + ['a negative elapsed time', { transfer: { ...transfer(), elapsedSeconds: -1 } }], + ['a nonfinite elapsed time', { transfer: { ...transfer(), elapsedSeconds: Number.NaN } }], + ['a negative phase elapsed time', { transfer: { ...transfer(), phaseElapsedSeconds: -1 } }], + ['a nonfinite phase elapsed time', { transfer: { ...transfer(), phaseElapsedSeconds: Number.POSITIVE_INFINITY } }], + ['a phase longer than the full transfer', { transfer: { ...transfer(), phaseElapsedSeconds: 11 } }], + ['a negative transfer rate', { transfer: { ...transfer(), bytesPerSecond: -1 } }], + ['a nonfinite transfer rate', { transfer: { ...transfer(), bytesPerSecond: Number.POSITIVE_INFINITY } }], + ['a negative remaining time', { transfer: { ...transfer(), etaSeconds: -1 } }], + ['a nonfinite remaining time', { transfer: { ...transfer(), etaSeconds: Number.POSITIVE_INFINITY } }] + ])('rejects %s without changing the existing cache', (_name, invalid) => { + libraries.updateHostLibraryCaches(hostId, [report()]); + const before = cache(); + vi.setSystemTime(Date.now() + 1_000); + expect(() => libraries.updateHostLibraryCaches(hostId, [report(invalid)])).toThrow(); + expect(cache()).toEqual(before); + }); + + it('rolls back the entire heartbeat when a later cache contains invalid metadata', async () => { + const second = await publish('t6'); + libraries.updateHostLibraryCaches(hostId, [report(), report({ libraryVersionId: second.id })]); + const beforeFirst = cache(); + const beforeSecond = libraries.getHostLibraryCache(hostId, second.id); + expect(() => libraries.updateHostLibraryCaches(hostId, [ + report({ bytesDownloaded: 70 }), + report({ libraryVersionId: second.id, transfer: { ...transfer(), elapsedSeconds: -1 } }) + ], { snapshot: true })).toThrow(); + expect(cache()).toEqual(beforeFirst); + expect(libraries.getHostLibraryCache(hostId, second.id)).toEqual(beforeSecond); + }); + + it('adds transfer columns to an existing database without losing cached files or operation references', () => { + const operation = libraries.queueLibraryDownload(hostId, version.id); + libraries.updateHostLibraryCaches(hostId, [{ + libraryVersionId: version.id, status: 'ready', manifestSha256: version.manifestSha256 + }]); + const before = cache(); + database.db().exec('ALTER TABLE host_library_caches DROP COLUMN transfer_progress'); + database.db().exec('ALTER TABLE host_library_caches DROP COLUMN bytes_verified'); + database.closeDb(); + expect(cache()).toMatchObject({ + hostId, libraryVersionId: version.id, status: 'ready', bytesDownloaded: 100, + bytesTotal: 100, manifestSha256: version.manifestSha256, + operationId: operation.id, updatedAt: before.updatedAt + }); + expect(cache().transfer).toBeUndefined(); + expect(libraries.requireReadyLibrary(hostId, version.id).id).toBe(version.id); + database.closeDb(); + expect(cache().operationId).toBe(operation.id); + }); +}); diff --git a/tests/unit/ui-game-file-transfer.test.ts b/tests/unit/ui-game-file-transfer.test.ts new file mode 100644 index 0000000..a539ca1 --- /dev/null +++ b/tests/unit/ui-game-file-transfer.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, it } from 'vitest'; +import type { HostLibraryCache } from '../../src/lib/game-files.js'; +import { gameFileTransferProgress } from '../../src/lib/ui/game-file-transfer.js'; + +const now = Date.parse('2026-10-04T16:00:00Z'); +function cache(changes: Partial = {}): HostLibraryCache { + return { + hostId: 'host-1', libraryVersionId: 't5-upload', status: 'downloading', bytesDownloaded: 200, bytesVerified: 0, bytesTotal: 1000, + manifestSha256: null, operationId: 'operation-1', error: null, updatedAt: new Date(now).toISOString(), + transfer: { phase: 'downloading', startedAt: new Date(now - 10_000).toISOString(), elapsedSeconds: 10, + phaseElapsedSeconds: 10, bytesPerSecond: 20, etaSeconds: 40 }, + ...changes + }; +} + +describe('game file transfer presentation', () => { + it('ticks elapsed time without inventing downloaded bytes or counting down an old estimate', () => { + expect(gameFileTransferProgress(cache(), true, now + 5000)).toMatchObject({ + phase: 'downloading', phaseBytes: 200, downloaded: 200, total: 1000, + elapsedSeconds: 15, bytesPerSecond: 20, etaSeconds: 40, live: true + }); + }); + + it('switches progress and the estimate to verification without resetting total elapsed', () => { + const verifying = cache({ status: 'verifying', bytesDownloaded: 1000, bytesVerified: 100, + transfer: { phase: 'verifying', startedAt: new Date(now - 60_000).toISOString(), elapsedSeconds: 60, + phaseElapsedSeconds: 2, bytesPerSecond: 50, etaSeconds: 18 } }); + expect(gameFileTransferProgress(verifying, true, now + 1000)).toMatchObject({ + phase: 'verifying', phaseBytes: 100, downloaded: 1000, elapsedSeconds: 61, bytesPerSecond: 50, etaSeconds: 18 + }); + }); + + it('freezes at the reported duration and hides estimates while offline', () => { + expect(gameFileTransferProgress(cache(), false, now + 12_000)).toMatchObject({ elapsedSeconds: 10, etaSeconds: null, bytesPerSecond: null, live: false }); + }); + + it('stops the clock and removes the estimate when reports become stale', () => { + expect(gameFileTransferProgress(cache(), true, now + 30_001)).toMatchObject({ elapsedSeconds: 40, etaSeconds: null, bytesPerSecond: null, stale: true }); + expect(gameFileTransferProgress(cache(), true, now + 600_000).elapsedSeconds).toBe(40); + }); + + it.each(['ready', 'outdated', 'failed'] as const)('keeps a final duration for %s copies', status => { + expect(gameFileTransferProgress(cache({ status }), true, now + 600_000)).toMatchObject({ + elapsedSeconds: 10, bytesPerSecond: null, etaSeconds: null, active: false, stale: false + }); + }); + + it('does not invent a timer for queued downloads or old agents', () => { + expect(gameFileTransferProgress(cache({ transfer: undefined }), true, now + 60_000)).toMatchObject({ elapsedSeconds: null, etaSeconds: null, phaseBytes: 200 }); + expect(gameFileTransferProgress(cache({ status: 'verifying', transfer: undefined, bytesVerified: undefined, bytesDownloaded: 1000 }), true, now)) + .toMatchObject({ elapsedSeconds: null, etaSeconds: null, phase: 'verifying', phaseBytes: null }); + }); + + it.each([0, null, Number.NaN, Number.POSITIVE_INFINITY, -1])('hides estimates for an unavailable transfer rate (%s)', bytesPerSecond => { + const current = cache(); + current.transfer!.bytesPerSecond = bytesPerSecond; + expect(gameFileTransferProgress(current, true, now)).toMatchObject({ etaSeconds: null, bytesPerSecond: null }); + }); + + it('does not show a misleading estimate before any bytes arrive or after all bytes arrive', () => { + for (const bytesDownloaded of [0, 1000]) { + expect(gameFileTransferProgress(cache({ bytesDownloaded }), true, now)).toMatchObject({ etaSeconds: null, bytesPerSecond: null }); + } + }); + + it('does not reuse a download estimate after verification starts', () => { + expect(gameFileTransferProgress(cache({ status: 'verifying', bytesDownloaded: 1000, bytesVerified: 50 }), true, now + 5000)) + .toMatchObject({ phase: 'verifying', phaseBytes: 50, etaSeconds: null, elapsedSeconds: 10 }); + }); + + it('handles an invalid report timestamp without advancing the timer or displaying estimates', () => { + expect(gameFileTransferProgress(cache({ updatedAt: 'invalid' }), true, now)).toMatchObject({ elapsedSeconds: 10, etaSeconds: null, stale: true }); + }); +});