Add persistent AFK reserves and recovery (Refs #151)

static/agent/ec-agent.py

@@ -35,7 +35,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.11.0"
+VERSION = "1.12.0"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -49,6 +49,7 @@ PLUTONIUM_MANIFEST_LIMIT = 2 * 1024 * 1024
 PLUTONIUM_FILE_LIMIT = 512 * 1024 * 1024
 PLUTONIUM_CACHE_LIMIT = 4 * 1024 * 1024 * 1024
 BANK_GAME_ENDPOINT = "https://ec.thewrightserver.net/api/bank"
+AFK_GAME_ENDPOINT = "https://ec.thewrightserver.net/api/afk"
 MODERATION_CONFIG_PATH = "configs/electric_cherry_moderation.cfg"
 MODERATION_INCLUDE = b"exec electric_cherry_moderation.cfg\n"
 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
@@ -1053,6 +1054,86 @@ class LibraryTransfer:
             self.persist(force=True)
 
 
+def afk_recovery_requests(journal):
+    """Mirror the native journal's deterministic recovery after a proven stop.
+
+    Active crashes never invent an elapsed duration. Their checkpoint is only
+    a lower bound; EC retains the hold for an audited correction. Requests that
+    may already have reached EC preserve their original identities and bodies.
+    """
+    def identifier(value):
+        return isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,64}", value)
+
+    def seconds(value):
+        return type(value) is int and 0 <= value <= 604800
+
+    def decode(raw):
+        if not isinstance(raw, str) or not 0 < len(raw.encode()) < 0x5000:
+            raise ValueError()
+        value = json.loads(raw)
+        if (not isinstance(value, dict) or value.get("server") != journal["server"]
+                or not all(identifier(value.get(key)) for key in ("guid", "match", "request_id"))
+                or value.get("action") not in ("event", "purchase", "reserve", "start", "settle")):
+            raise ValueError()
+        if value["action"] in ("reserve", "start", "settle") and not identifier(value.get("session_id")):
+            raise ValueError()
+        if value["action"] == "reserve" and (not seconds(value.get("seconds")) or value["seconds"] == 0):
+            raise ValueError()
+        return value
+
+    try:
+        if (not isinstance(journal, dict) or type(journal.get("version")) is not int
+                or journal["version"] != 1 or not isinstance(journal.get("server"), str)
+                or not 0 < len(journal["server"]) <= 128
+                or journal.get("endpoint") != AFK_GAME_ENDPOINT
+                or not isinstance(journal.get("requests"), list)
+                or not isinstance(journal.get("sessions"), list)
+                or len(journal["requests"]) + len(journal["sessions"]) > 4096):
+            raise ValueError()
+        requests = {}
+        for raw in journal["requests"]:
+            value = decode(raw)
+            if value["request_id"] in requests:
+                raise ValueError()
+            requests[value["request_id"]] = value
+        sessions = set()
+        for session in journal["sessions"]:
+            data = decode(session["body"])
+            session_id = data.get("session_id")
+            phase = session["phase"]
+            authorized, consumed = session["authorized_seconds"], session["consumed_seconds"]
+            if (data["action"] != "reserve" or session_id in sessions
+                    or phase not in ("prepared", "authorized", "active")
+                    or not seconds(authorized) or not seconds(consumed)
+                    or consumed > authorized or authorized > data["seconds"]
+                    or (phase == "prepared" and authorized != 0)
+                    or (phase != "prepared" and authorized == 0)
+                    or (phase != "active" and consumed != 0)):
+                raise ValueError()
+            sessions.add(session_id)
+            requests = {key: value for key, value in requests.items()
+                        if not (value["action"] in ("reserve", "start") and value["session_id"] == session_id)}
+            hashed = 14695981039346656037
+            scope = journal["server"] + "\0" + journal["endpoint"] + "\0" + session_id
+            for byte in scope.encode():
+                hashed = ((hashed ^ byte) * 1099511628211) & 0xffffffffffffffff
+            transfer = {"server": journal["server"], "guid": data["guid"], "match": data["match"],
+                        "session_id": session_id, "request_id": f"afk-stop-{hashed:016x}",
+                        "action": "settle", "end_reason": "crash" if phase == "active" else "never_started",
+                        "consumed_seconds": None if phase == "active" else 0}
+            if phase == "active":
+                transfer["confirmed_seconds"] = consumed
+            previous = requests.get(transfer["request_id"])
+            if previous is not None and previous != transfer:
+                raise ValueError()
+            requests[transfer["request_id"]] = transfer
+        if any(value["action"] in ("reserve", "start") for value in requests.values()):
+            raise ValueError()
+        return list(requests.values())
+    except (ValueError, TypeError, KeyError, AttributeError):
+        raise AgentError("AFK journal is invalid; runtime storage retained") from None
+
+
 class Agent:
     def __init__(self, config, docker=None):
         self.config = config
@@ -1574,7 +1655,7 @@ class Agent:
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["host.temperatures", "library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.rebuild-files", "instance.autoscaling-remove", "moderation.configure", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset", "bank.settlement"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["host.temperatures", "library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.rebuild-files", "instance.autoscaling-remove", "moderation.configure", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset", "bank.settlement", "afk.settlement"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
         plutonium_cache = self.plutonium_cache_inventory()
         return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "temperatures": temperatures, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "preparedRuntimeImages": sorted(set(prepared_images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
@@ -2237,6 +2318,52 @@ class Agent:
                         raise AgentError("Bank settlement was not acknowledged; runtime storage retained")
             except (ValueError, TypeError, KeyError, AttributeError):
                 raise AgentError("Bank journal is invalid; runtime storage retained") from None
+        self.settle_afk(instance_id, generation, operation_id)
+
+    def settle_afk(self, instance_id, generation, operation_id=None):
+        """Retain AFK evidence until stopped-process reconciliation is acknowledged."""
+        app = scoped(self.instance_dir(instance_id, generation), "app")
+        if not app.exists():
+            return
+        if self.inspect(instance_id, generation)["running"]:
+            raise AgentError("AFK settlement requires a stopped instance")
+        operation_id = operation_id or self.executing_id
+        visited, journals = 0, []
+        deadline = time.monotonic() + 30
+
+        def unreadable(_error):
+            raise AgentError("AFK journal discovery failed; runtime storage retained")
+
+        for directory, directories, files in os.walk(app, followlinks=False, onerror=unreadable):
+            relative = Path(directory).relative_to(app)
+            inside = "tws-afk" in relative.parts
+            visited += 1
+            if visited > 100000 or time.monotonic() > deadline:
+                raise AgentError("AFK journal discovery limit exceeded; runtime storage retained")
+            for name in directories + files:
+                if (inside or name == "tws-afk") and (Path(directory) / name).is_symlink():
+                    raise AgentError("AFK journal paths must not be symlinks")
+            if inside:
+                journals.extend((relative / name).as_posix() for name in files if name.endswith(".json"))
+                if len(journals) > 4096:
+                    raise AgentError("Too many AFK journals; runtime storage retained")
+        for path in sorted(journals):
+            try:
+                raw = managed_read(app, path)
+                if raw is None or len(raw) > 16 * 1024 * 1024:
+                    raise ValueError()
+                transfers = afk_recovery_requests(json.loads(raw))
+            except (Rejected, ValueError, UnicodeError):
+                raise AgentError("AFK journal cannot be read; runtime storage retained") from None
+            if transfers and not operation_id:
+                raise AgentError("AFK settlement requires a teardown operation")
+            for transfer in transfers:
+                response = self.request("POST", "/api/agent/afk/settle", {
+                    "operationId": operation_id, "action": transfer["action"], "transfer": transfer})
+                expected = ("settled", "unresolved", "corrected") if transfer["action"] == "settle" else ("rewarded", "purchased", "rejected")
+                if (not isinstance(response, dict) or response.get("request_id") != transfer["request_id"]
+                        or response.get("state") not in expected):
+                    raise AgentError("AFK recovery was not acknowledged; runtime storage retained")
 
     def lifecycle(self, kind, instance_id, generation, payload):
         unhealthy_recovery = payload.get("unhealthyRecovery") is True

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.11.0",
+  "version": "1.12.0",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "3cb08d5199a6a3bbdc5641e17a7f752b65c79935813f299ea245ffe449a4e861",
+  "sha256": "7b8212ea4a57ef7a37ee3916a0ec0e19965928e373304a64645f3998f87d51d7",
   "updaterProtocolVersion": 2,
   "updaterSha256": "a9d7f96b87d0ec74d8a0f476dff6e3d2a7a0bc454a93a51d7b57de00c9ce23cb",
   "installerSha256": "384d5e397a51d19cb9fc9ed8793b4314353f726e217030246050859c0bc18231"

static/agent/manifest.json.sig

@@ -1,8 +0,0 @@
------BEGIN PGP SIGNATURE-----
-
-iJEEABYKADkWIQSpocb6VBfQzvdIh9HF6fdSNniN+gUCatJ0WhsUgAAAAAAEAA5t
-YW51MiwyLjUrMS4xMiwwLDMACgkQxen3UjZ4jfqjWQD8CaV5ZH/kD1AhJ15B9czG
-x69WnENeMXHMjHt0vnIlSuQBAKE+r6scq6M5Y2rbUDbgDFDjir5lR36qR8DGBi6A
-VMQH
-=V8dS
------END PGP SIGNATURE-----