From 5455f27d5bbf0c26d7ce77f63dfbb49ac289282e Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 5 Oct 2026 15:55:08 -0400 Subject: [PATCH] Cache agent inventory manifests by modification time --- agent/ec_agent.py | 36 ++++++++++++++++--- agent/tests/test_agent.py | 73 ++++++++++++++++++++++++++++++++++++++ docs/native-hosts.md | 4 +++ static/agent/ec-agent.py | 36 ++++++++++++++++--- static/agent/manifest.json | 4 +-- 5 files changed, 141 insertions(+), 12 deletions(-) diff --git a/agent/ec_agent.py b/agent/ec_agent.py index f2b2d58..f195d92 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.10" +VERSION = "1.4.11" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -918,6 +918,8 @@ class Agent: self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.library_progress = {} self.library_transfers = {} + self.inventory_manifest_cache = {} + self.inventory_manifest_lock = threading.Lock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None @@ -1191,6 +1193,25 @@ class Agent: continue return samples + def inventory_manifest(self, marker): + # Inventory only: launch/download verification must still read disk. + # Include file identity so atomic replacements invalidate even when + # their size and modification time match the previous manifest. + with self.inventory_manifest_lock: + try: + metadata = marker.stat() + signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino) + cached = self.inventory_manifest_cache.get(marker) + if cached is not None and cached[0] == signature: + return cached[1] + self.inventory_manifest_cache.pop(marker, None) + value = json.loads(marker.read_text()) + self.inventory_manifest_cache[marker] = (signature, value) + return value + except (OSError, ValueError): + self.inventory_manifest_cache.pop(marker, None) + raise + def inventory(self): cpu_usage = self.cpu_usage_percent() usage = shutil.disk_usage(self.root) @@ -1210,13 +1231,18 @@ class Agent: 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"): + manifests = list((self.root / "libraries").glob("*/*/manifest.json")) + partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json")) + with self.inventory_manifest_lock: + for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests): + del self.inventory_manifest_cache[marker] + for marker in manifests: # A crash between writing a manifest and its atomic promotion must # never expose the staging directory as a ready library. if marker.parent.name.startswith("."): continue try: - value = json.loads(marker.read_text()) + value = self.inventory_manifest(marker) # 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(): @@ -1233,9 +1259,9 @@ class Agent: except (OSError, ValueError, KeyError): continue ready = {item["id"] for item in libraries} - for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): + for marker in partial_manifests: try: - value = json.loads(marker.read_text()) + value = self.inventory_manifest(marker) library_id = identifier(value["libraryId"]) if library_id in ready or library_id in progress: continue diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 1f1b122..3f5c1db 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -386,6 +386,68 @@ class AgentTests(unittest.TestCase): self.agent.journal.close() self.temp.cleanup() + def test_inventory_reuses_manifests_until_mtime_changes(self): + marker = self.library / "manifest.json" + saved = json.loads(marker.read_text()) + original_read = Path.read_text + reads = [] + + def read(path, *args, **kwargs): + if path == marker: + reads.append(path) + return original_read(path, *args, **kwargs) + + with patch.object(Path, "read_text", read): + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9) + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9) + self.assertEqual(len(reads), 1) + metadata = marker.stat() + marker.write_text(json.dumps({**saved, "size": 8})) + module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000)) + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) + self.assertEqual(len(reads), 2) + + def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self): + marker = self.library / "manifest.json" + saved = json.loads(marker.read_text()) + self.agent.inventory() + metadata = marker.stat() + module.atomic_json(marker, {**saved, "size": 8}) + module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns)) + self.assertEqual(marker.stat().st_size, metadata.st_size) + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) + + def test_inventory_discards_invalid_and_removed_manifests(self): + marker = self.library / "manifest.json" + saved = json.loads(marker.read_text()) + self.agent.inventory() + marker.write_text("invalid JSON") + self.assertEqual(self.agent.inventory()["libraryCaches"], []) + self.assertNotIn(marker, self.agent.inventory_manifest_cache) + module.atomic_json(marker, saved) + self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1) + marker.unlink() + self.assertEqual(self.agent.inventory()["libraryCaches"], []) + self.assertNotIn(marker, self.agent.inventory_manifest_cache) + module.atomic_json(marker, {**saved, "size": 8}) + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) + + def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self): + self.agent.inventory() + progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3} + module.atomic_json(self.library / "transfer.json", progress) + self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing") + module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5}) + cache = self.agent.inventory()["libraryCaches"][0] + self.assertEqual(cache["status"], "failed") + self.assertEqual(cache["bytesVerified"], 5) + (self.library / "content/game.ff").write_bytes(b"bad-data!") + with self.assertRaises(AgentError): + self.agent.verify_library(self.library) + (self.library / "manifest.json").write_text("invalid JSON") + with self.assertRaises(ValueError): + self.agent.verify_library(self.library) + def test_library_remove_preserves_other_versions_and_is_idempotent(self): other = self.agent.library_dir("t5", "other-library") other.mkdir() @@ -1179,6 +1241,17 @@ class AgentTests(unittest.TestCase): self.assertEqual(caches[0]["status"], "missing") self.assertEqual(caches[0]["bytesDownloaded"], len(content)) self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"]) + (staging / "content/zone/test.ff").write_bytes(content[:5]) + original_read = Path.read_text + + def read(path, *args, **kwargs): + if path == staging / "source.json": + self.fail("Unchanged partial manifest should be cached") + return original_read(path, *args, **kwargs) + + with patch.object(Path, "read_text", read): + partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") + self.assertEqual(partial["bytesDownloaded"], 5) self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3} caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"] self.assertEqual(len(caches), 1) diff --git a/docs/native-hosts.md b/docs/native-hosts.md index 77b358d..ac1d05c 100644 --- a/docs/native-hosts.md +++ b/docs/native-hosts.md @@ -595,6 +595,10 @@ missing, stale, or predates lifecycle changes. RCON status queries share one operation per host, with two concurrent local probes and a 20-second batch deadline. Individual failures remain visible per server. Both additions are capability-gated; older agents retain individual inspection and RCON operations. +Agent 1.4.11 caches parsed library manifests during inventory collection until +their modification time, size, or file identity changes. Deleted manifests leave +the cache, while transfer progress and partial-file sizes are sampled each time. +Launch and download integrity verification continue to read from disk. Agent 1.4.5 reserves two observation workers for instance health and read-only RCON, and one for files, logs, and host inventory. Background reads arrive in batches of up to four. File receipt uploads release the instance lock before contacting the diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index f2b2d58..f195d92 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.10" +VERSION = "1.4.11" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -918,6 +918,8 @@ class Agent: self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.library_progress = {} self.library_transfers = {} + self.inventory_manifest_cache = {} + self.inventory_manifest_lock = threading.Lock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None @@ -1191,6 +1193,25 @@ class Agent: continue return samples + def inventory_manifest(self, marker): + # Inventory only: launch/download verification must still read disk. + # Include file identity so atomic replacements invalidate even when + # their size and modification time match the previous manifest. + with self.inventory_manifest_lock: + try: + metadata = marker.stat() + signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino) + cached = self.inventory_manifest_cache.get(marker) + if cached is not None and cached[0] == signature: + return cached[1] + self.inventory_manifest_cache.pop(marker, None) + value = json.loads(marker.read_text()) + self.inventory_manifest_cache[marker] = (signature, value) + return value + except (OSError, ValueError): + self.inventory_manifest_cache.pop(marker, None) + raise + def inventory(self): cpu_usage = self.cpu_usage_percent() usage = shutil.disk_usage(self.root) @@ -1210,13 +1231,18 @@ class Agent: 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"): + manifests = list((self.root / "libraries").glob("*/*/manifest.json")) + partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json")) + with self.inventory_manifest_lock: + for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests): + del self.inventory_manifest_cache[marker] + for marker in manifests: # A crash between writing a manifest and its atomic promotion must # never expose the staging directory as a ready library. if marker.parent.name.startswith("."): continue try: - value = json.loads(marker.read_text()) + value = self.inventory_manifest(marker) # 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(): @@ -1233,9 +1259,9 @@ class Agent: except (OSError, ValueError, KeyError): continue ready = {item["id"] for item in libraries} - for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): + for marker in partial_manifests: try: - value = json.loads(marker.read_text()) + value = self.inventory_manifest(marker) library_id = identifier(value["libraryId"]) if library_id in ready or library_id in progress: continue diff --git a/static/agent/manifest.json b/static/agent/manifest.json index d296e7f..06e1487 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.4.10", + "version": "1.4.11", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "c518a216ad4e9dd86ac82743b27bae54cdf1c51ef7ae3bf8d252dcc9863be1fb", + "sha256": "4c6cc180f3c45a436878a0278d62b6840526150e039fdf51bafadef63759b131", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" }