josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Add cancellation and cleanup for native server creation

Browse Source
This commit is contained in:
josh
2026-10-04 18:00:11 -04:00
parent 19f1f4e279
commit 176b7dd86b
17 changed files with 482 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/ec_agent.py
+57 -3
View File
@@ -30,7 +30,7 @@ import urllib.parse
import urllib.request
import uuid
VERSION = "1.2.2"
VERSION = "1.3.0"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
LIBRARY_DOWNLOAD_WORKERS = 4
@@ -672,6 +672,7 @@ class Agent:
self.docker = docker or Docker()
self.stopping = threading.Event()
self.current_operation = None
self.cancelled_creations = frozenset()
self.library_progress = {}
self.library_transfers = {}
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
@@ -952,7 +953,7 @@ class Agent:
if line.startswith("PRETTY_NAME="):
os_name = line.split("=", 1)[1].strip('"')
update = self.update_inventory()
capabilities = (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
capabilities = ["instance.cancel-create"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
def safe_to_disrupt(self, instance_id, generation):
@@ -1040,6 +1041,7 @@ class Agent:
return game, library, image, port, bind, env
def create(self, instance_id, generation, payload):
self.check_creation_cancellation()
directory = self.instance_dir(instance_id, generation)
current_marker = directory.parent / "current.json"
if current_marker.exists():
@@ -1068,12 +1070,19 @@ class Agent:
for protocol in ("udp", "tcp"):
args.extend(["--publish", f"{bind_arg}:{port}:{port}/{protocol}"])
args.append(image)
self.check_creation_cancellation()
self.checkpoint({"phase": "creating", "generation": generation, "specSha256": spec_sha})
self.docker.call(*args, timeout=120)
self.check_creation_cancellation()
self.docker.call("container", "start", name)
self.check_creation_cancellation()
atomic_json(current_marker, {"generation": generation})
return self.inspect(instance_id, generation)
def check_creation_cancellation(self):
if self.executing_id and self.executing_id in self.cancelled_creations:
raise AgentError("Creation cancelled; waiting for instance cleanup")
def write_initial_files(self, root, files):
if not isinstance(files, list) or len(files) > 10000:
raise AgentError("Invalid file list")
@@ -1093,6 +1102,8 @@ class Agent:
return content
def lifecycle(self, kind, instance_id, generation, payload):
if kind == "instance.cancel-create":
return self.cancel_creation(instance_id, generation, payload)
if kind == "instance.create":
return self.create(instance_id, generation, payload)
if kind == "instance.inspect":
@@ -1146,6 +1157,38 @@ class Agent:
raise AgentError("Unsupported instance operation")
return {**self.inspect(instance_id, generation), "before": before}
def cancel_creation(self, instance_id, generation, payload):
"""Explicit launch cancellation, fenced to its immutable creation identity.
The operation loop serializes this with create. Once delivered, removal
needs no RCON response and a durable tombstone prevents a delayed create.
Repeating this cleanup after an ambiguous Docker result is safe.
"""
create_id = identifier(payload.get("createOperationId"))
marker = self.instance_dir(instance_id, generation).parent / "current.json"
if marker.exists() and json.loads(marker.read_text())["generation"] > generation:
raise Rejected("Cancelled creation was superseded by a newer generation")
saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (create_id,)).fetchone()
if saved:
original = json.loads(saved["request"])
if original["kind"] != "instance.create" or original["instanceId"] != instance_id or original["generation"] != generation:
raise Rejected("Cancellation does not match the original creation")
health = self.inspect(instance_id, generation)
if health["exists"] and health.get("creationOperation") != create_id:
raise Rejected("Container belongs to a different creation operation")
self.checkpoint({"phase": "cancelling-creation", "createOperationId": create_id})
# Fence before deleting: a crash must never permit the old creation to run again.
self.mark_removed(instance_id, generation)
if health["exists"]:
self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, generation))
after = self.inspect(instance_id, generation)
if after["exists"]:
raise AgentError("Cancelled creation container still exists")
directory = self.instance_dir(instance_id, generation)
if directory.exists():
shutil.rmtree(directory)
return {**after, "cancelledCreation": create_id}
def stage(self, instance_id, generation, payload, operation_id):
directory = self.instance_dir(instance_id, generation)
self.spec(instance_id, generation)
@@ -1609,7 +1652,7 @@ class Agent:
current_marker = self.instance_dir(instance_id, generation).parent / "current.json"
if current_marker.exists():
current = json.loads(current_marker.read_text())
if current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove"):
if current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove", "instance.cancel-create"):
raise AgentError("Instance generation has been superseded or removed")
if kind.startswith("instance."):
return self.lifecycle(kind, instance_id, generation, payload)
@@ -1673,6 +1716,14 @@ class Agent:
instance_id, generation = operation.get("instanceId"), operation.get("generation")
checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {}
try:
if kind == "instance.cancel-create":
return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload))
if kind == "instance.create":
marker = self.instance_dir(instance_id, generation).parent / "current.json"
if marker.exists():
current = json.loads(marker.read_text())
if current["generation"] > generation or current["generation"] == generation and current.get("removed"):
return self.journal.update(operation["id"], "failed", error="Creation was cancelled or superseded")
if kind == "host.decommission":
self.executing_id = operation["id"]
try:
@@ -1792,6 +1843,9 @@ class Agent:
def heartbeat(self):
response = self.request("POST", "/api/agent/heartbeat", {"inventory": self.inventory(), "operationId": self.current_operation})
# Only the heartbeat thread replaces this immutable snapshot; it never
# mutates the operation journal or races Docker cleanup against create.
self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str))
host = response.get("host", {})
if isinstance(host.get("portMin"), int) and isinstance(host.get("portMax"), int):
self.config["portRange"] = {"min": host["portMin"], "max": host["portMax"]}
agent/tests/test_agent.py
+91
View File
@@ -209,6 +209,97 @@ class AgentTests(unittest.TestCase):
self.assertTrue((self.library / "content/game.ff").exists())
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
def test_cancel_creation_bypasses_occupancy_and_preserves_library(self):
self.create()
self.agent.rcon = lambda *_: self.fail("Cancellation must not wait for player evidence")
operation = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
self.assertEqual(self.agent.execute(operation)["state"], "completed")
self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
self.assertTrue((self.library / "content/game.ff").exists())
self.assertEqual(self.agent.execute(operation)["state"], "completed")
self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1)
def test_heartbeat_cancellation_stops_create_before_starting_container(self):
original_call = self.docker.call
def cancel_during_create(*args, **kwargs):
result = original_call(*args, **kwargs)
if args[:2] == ("container", "create"):
self.agent.cancelled_creations = frozenset(["create"])
return result
self.docker.call = cancel_during_create
payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
self.assertEqual(self.agent.execute(self.operation("instance.create", payload, "create"))["state"], "unknown")
self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls))
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
self.assertEqual(result["state"], "completed")
self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])
def test_cancellation_cleanup_resumes_after_agent_restart(self):
self.create()
self.docker.lose_reply_after = "rm"
cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
self.assertEqual(self.agent.execute(cancel)["state"], "unknown")
self.agent.journal.db.close()
self.agent = Agent(self.config, self.docker)
self.assertEqual(self.agent.execute(cancel)["state"], "completed")
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
def test_cancel_before_creation_delivery_fences_delayed_create(self):
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "late-create"}))
self.assertEqual(result["state"], "completed", result)
result = self.agent.execute(self.operation("instance.create", {}, "late-create"))
self.assertEqual(result["state"], "failed")
self.assertFalse(any(call[:2] == ("container", "create") for call in self.docker.calls))
def test_cancel_removes_partial_files_without_a_container(self):
directory = self.agent.instance_dir(INSTANCE, 1)
(directory / "app").mkdir(parents=True)
(directory / "app/partial.cfg").write_text("partial")
self.agent.journal.accept(self.operation("instance.create", {}, "create"))
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
self.assertEqual(result["state"], "completed", result)
self.assertFalse(directory.exists())
self.assertTrue((self.library / "content/game.ff").exists())
def test_cancel_rejects_a_creation_id_from_another_instance(self):
other = self.operation("instance.create", {}, "other-create")
other["instanceId"] = "another-instance"
self.agent.journal.accept(other)
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "other-create"}))
self.assertEqual(result["state"], "failed", result)
self.assertFalse((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").exists())
def test_cancel_creation_rejects_other_creation_and_newer_generation(self):
self.create()
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "wrong"}))
self.assertEqual(result["state"], "failed")
module.atomic_json(self.agent.instance_dir(INSTANCE, 1).parent / "current.json", {"generation": 2})
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}, "old-generation"))
self.assertEqual(result["state"], "failed")
self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"])
self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls))
def test_cancel_creation_reconciles_lost_removal_reply_and_fences_unknown_create(self):
self.docker.lose_reply_after = "start"
payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
create = self.operation("instance.create", payload, "create")
self.assertEqual(self.agent.execute(create)["state"], "unknown")
self.docker.lose_reply_after = "rm"
cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
self.assertEqual(self.agent.execute(cancel)["state"], "unknown")
self.assertEqual(self.agent.execute(cancel)["state"], "completed")
self.assertEqual(self.agent.execute(create)["state"], "failed")
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
self.assertTrue(json.loads((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").read_text())["removed"])
def test_cancel_creation_rejects_unowned_container(self):
self.create()
self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {}
result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
self.assertEqual(result["state"], "failed")
self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls))
def test_unowned_container_never_mutated(self):
self.create()
self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {}