diff --git a/agent/ec_agent.py b/agent/ec_agent.py index c48f6ae..22f2a18 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.5.2" +VERSION = "1.5.3" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -41,6 +41,7 @@ MANIFEST_MAX_BYTES = 16 * 1024 * 1024 MANIFEST_MAX_SECONDS = 15 MANIFEST_MAX_DEPTH = 64 LIBRARY_DOWNLOAD_WORKERS = 4 +HEALTH_LOCK_TIMEOUT_SECONDS = 5 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") AGENT_CONFIG = Path("/etc/electric-cherry/agent.json") AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py") @@ -923,6 +924,28 @@ class Agent: self.inventory_manifest_lock = threading.Lock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None + self.lane_lock = threading.Lock() + self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False} + for name in ("health-1", "health-2", "read", "write")} + + def lane_started(self, name): + with self.lane_lock: + self.lanes[name]["thread"] = threading.current_thread() + + def lane_polled(self, name): + with self.lane_lock: + self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic()) + + def lane_busy(self, name, busy): + with self.lane_lock: + self.lanes[name]["busy"] = busy + + def lane_inventory(self): + with self.lane_lock: + return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(), + "busy": lane["busy"], "lastPollAt": lane["lastPollAt"], + "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])} + for name, lane in self.lanes.items()] def can_uninstall(self): try: @@ -1334,7 +1357,7 @@ class Agent: os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest"] + (["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, **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)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} + 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, **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)), "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): health = self.inspect(instance_id, generation) @@ -2366,13 +2389,29 @@ class Agent: key = operation.get("instanceId") return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext() + @contextlib.contextmanager + def observation_lock(self, operation, health=False): + lock = self.instance_lock(operation) + if not health or not operation.get("instanceId"): + with lock: + yield + return + if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS): + raise AgentError("Health check timed out waiting for the instance operation") + try: + yield + finally: + lock.release() + @staticmethod def health_operation(operation): return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version")) - def read_loop(self, lane="read"): + def read_loop(self, lane="read", lane_name=None): # Each lane owns its SQLite connection and never writes executing_id, # checkpoints, or current_operation used by the mutation state machine. + lane_name = lane_name or ("health-1" if lane == "health" else "read") + self.lane_started(lane_name) journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False) next_prune = time.monotonic() try: @@ -2388,7 +2427,9 @@ class Agent: if journal.prune_observations(self.observation_keep_days) == 500: next_prune = time.monotonic() + 15 response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20") + self.lane_polled(lane_name) handled = False + self.lane_busy(lane_name, bool(response.get("operations"))) for operation in response.get("operations", []): if self.stopping.is_set() or not self.read_operation(operation): continue @@ -2405,12 +2446,12 @@ class Agent: row, _ = journal.accept(operation) if row["state"] in ("accepted", "unknown", "running"): self.report(journal.update(operation["id"], "running"), journal=journal) - with self.read_gate.read(), self.instance_lock(operation): - try: + try: + with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"): result = self.dispatch(operation) row = journal.update(operation["id"], "completed", result) - except Exception as err: - row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") + except Exception as err: + row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") # Receipt delivery can take seconds; it must not # hold the instance lock needed by health checks. self.report(row, journal=journal) @@ -2427,6 +2468,8 @@ class Agent: except Exception as err: print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr) self.stopping.wait(3) + finally: + self.lane_busy(lane_name, False) finally: journal.close() @@ -2440,6 +2483,8 @@ class Agent: return # Never hold the updater lock while waiting for delivery. response = self.request("GET", "/api/agent/operations?lane=write&wait=20") + self.lane_polled("write") + self.lane_busy("write", bool(response.get("operations"))) with self.activity() as acquired: if not acquired or self.stopping.is_set() or self.update_pending(): return @@ -2467,6 +2512,7 @@ class Agent: return completed def run(self): + self.lane_started("write") # Obtain the configured host port range before accepting any deployment. # Enrollment's default values must never race the first heartbeat. while not self.stopping.is_set(): @@ -2487,7 +2533,8 @@ class Agent: heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True) heartbeats.start() # Preserve two concurrent health checks even while files are being read. - readers = [threading.Thread(target=self.read_loop, args=(lane,), daemon=True) for lane in ("health", "health", "read")] + readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True) + for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))] for reader in readers: reader.start() while not self.stopping.is_set(): @@ -2499,6 +2546,8 @@ class Agent: print("Host controller contact unavailable; existing containers continue running", file=sys.stderr) except Exception as err: print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr) + finally: + self.lane_busy("write", False) self.stopping.wait(0.25 if completed else 3) for reader in readers: reader.join(timeout=2) diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 223373f..06a483f 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -446,6 +446,67 @@ class AgentTests(unittest.TestCase): self.assertEqual(logs.getvalue().count("RuntimeError"), 2) self.assertNotIn("secret", logs.getvalue()) + def test_lane_inventory_tracks_each_health_worker_and_detects_dead_threads(self): + def poll(): + self.agent.lane_started("health-1") + self.agent.lane_polled("health-1") + + with patch.object(module.time, "monotonic", return_value=100): + thread = threading.Thread(target=poll) + thread.start() + thread.join(timeout=2) + self.assertFalse(thread.is_alive()) + with patch.object(module.time, "monotonic", return_value=150): + self.agent.lane_started("health-2") + self.agent.lane_polled("health-2") + self.agent.lane_busy("health-2", True) + with patch.object(module.time, "monotonic", return_value=200): + lanes = {lane["name"]: lane for lane in self.agent.lane_inventory()} + self.assertEqual(set(lanes), {"health-1", "health-2", "read", "write"}) + self.assertFalse(lanes["health-1"]["alive"]) + self.assertEqual(lanes["health-1"]["pollAgeSeconds"], 100) + self.assertTrue(lanes["health-2"]["alive"]) + self.assertTrue(lanes["health-2"]["busy"]) + self.assertEqual(lanes["health-2"]["pollAgeSeconds"], 50) + self.assertIsNotNone(lanes["health-2"]["lastPollAt"]) + self.assertEqual(len(self.agent.inventory()["agentLanes"]), 4) + + def test_health_lane_times_out_on_instance_lock_and_processes_next_check(self): + blocked = self.operation("instance.inspect", op_id="blocked-health") + fresh = {**self.operation("instance.inspect", op_id="fresh-health"), "instanceId": "other-instance"} + pending = iter([blocked, fresh]) + failures = [] + + def request(method, path, payload=None): + if method == "POST": + return {"operation": {"id": path.rsplit("/", 1)[1], "state": payload["state"]}} + operation = next(pending, None) + if operation is None: + self.agent.stopping.set() + return {"operations": [operation] if operation else []} + + def read(): + try: + self.agent.read_loop("health") + except BaseException as error: + failures.append(error) + + with patch.object(module, "HEALTH_LOCK_TIMEOUT_SECONDS", 0.01), patch.object(self.agent, "request", side_effect=request), patch.object(self.agent, "dispatch", return_value={"healthy": True}) as dispatch: + with self.agent.instance_lock(blocked): + thread = threading.Thread(target=read) + thread.start() + thread.join(timeout=2) + finished_while_locked = not thread.is_alive() + self.agent.stopping.set() + thread.join(timeout=2) + self.assertTrue(finished_while_locked) + self.assertEqual(failures, []) + self.assertEqual(self.agent.journal.get(blocked["id"])["state"], "failed") + self.assertIn("waiting for the instance", self.agent.journal.get(blocked["id"])["error"]) + self.assertEqual(self.agent.journal.get(fresh["id"])["state"], "completed") + dispatch.assert_called_once_with(fresh) + self.assertEqual(self.agent.read_claims, set()) + def test_inventory_reuses_manifests_until_mtime_changes(self): marker = self.library / "manifest.json" saved = json.loads(marker.read_text()) diff --git a/src/lib/hosts.ts b/src/lib/hosts.ts index c7e572a..af77af1 100644 --- a/src/lib/hosts.ts +++ b/src/lib/hosts.ts @@ -35,6 +35,21 @@ export interface HostAgentUpdate { checkedAt?: string; } +export interface HostAgentLane { + name: 'health-1' | 'health-2' | 'read' | 'write'; + alive: boolean; busy: boolean; lastPollAt: string | null; pollAgeSeconds: number; +} +export function hostAgentLaneLabel(name: HostAgentLane['name']): string { + return { 'health-1': 'Health check 1', 'health-2': 'Health check 2', read: 'File reads', write: 'Actions' }[name]; +} +export function hostAgentLaneState(lane: HostAgentLane): 'stopped' | 'stalled' | 'busy' | 'starting' | 'polling' { + if (!lane.alive) return 'stopped'; + // Reads and mutations can legitimately take minutes. Health work has a + // bounded budget; an old poll there indicates a blocked worker even if busy. + if (lane.pollAgeSeconds > 90 && (!lane.busy || lane.name.startsWith('health-'))) return 'stalled'; + return lane.busy ? 'busy' : lane.lastPollAt === null ? 'starting' : 'polling'; +} + export function hostAgentUpdatesEnabled(inventory: HostInventory): boolean { return inventory.agentUpdate?.enabled === true && inventory.capabilities?.includes('agent.update') === true; } @@ -68,6 +83,7 @@ export interface HostInventory { currentOperation?: string | null; [key: string]: unknown; capabilities?: string[]; agentUpdate?: HostAgentUpdate; + agentLanes?: HostAgentLane[]; instanceResources?: InstanceResources[]; instanceResourcesError?: string; instanceStates?: InstanceState[]; diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index 6ea1639..ee12437 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -11,6 +11,7 @@ import { parseHostInventory } from './request-context.js'; import { fleetSettings } from '../settings.js'; import { recordInstanceResources, validateInstanceResources } from './calibration.js'; import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js'; +import { hostAgentLaneLabel, hostAgentLaneState } from '../../hosts.js'; import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationFilter, HostOperationHistoryPage, HostOperationScope, HostOperationState } from '../../hosts.js'; export const AGENT_PROTOCOL_VERSION = 1; @@ -46,6 +47,10 @@ function hostFromRow(row: HostRow): Host { if (!row.enrolled_at) reasons.push('Agent has not enrolled.'); else if (!row.credential_active) reasons.push('Agent credential has been revoked.'); if (!connected) reasons.push('Agent heartbeat is unavailable or stale.'); + if (connected) for (const lane of inventory.agentLanes ?? []) { + const state = hostAgentLaneState(lane); + if (state === 'stopped' || state === 'stalled') reasons.push(`Agent ${hostAgentLaneLabel(lane.name).toLowerCase()} worker is ${state}. Check the host agent service.`); + } if (inventory.protocolVersion !== AGENT_PROTOCOL_VERSION) reasons.push('A compatible agent protocol is required.'); if (!inventory.dockerRunning) reasons.push('Docker is not ready.'); if (!inventory.directoriesWritable) reasons.push('Managed directories are not writable.'); @@ -287,6 +292,20 @@ export function validateInventory(value: unknown): HostInventory { ...(update.error === undefined ? {} : { error: update.error }), ...(update.checkedAt === undefined ? {} : { checkedAt: update.checkedAt }) }; } + if (inventory.agentLanes !== undefined) { + if (!Array.isArray(inventory.agentLanes) || inventory.agentLanes.length > 4) throw new Error('Invalid agent lane inventory.'); + // Keep valid lane diagnostics without discarding a host heartbeat over + // one malformed sample. Do not persist thread details or raw errors. + const names = new Set(); + inventory.agentLanes = inventory.agentLanes.filter(lane => { + if (!lane || !['health-1','health-2','read','write'].includes(lane.name) || names.has(lane.name) + || typeof lane.alive !== 'boolean' || typeof lane.busy !== 'boolean' + || typeof lane.pollAgeSeconds !== 'number' || !Number.isFinite(lane.pollAgeSeconds) || lane.pollAgeSeconds < 0 + || (lane.lastPollAt !== null && (typeof lane.lastPollAt !== 'string' || lane.lastPollAt.length > 40 || !Number.isFinite(Date.parse(lane.lastPollAt))))) return false; + names.add(lane.name); + return true; + }).map(({name,alive,busy,lastPollAt,pollAgeSeconds}) => ({name,alive,busy,lastPollAt,pollAgeSeconds})); + } if (inventory.libraries !== undefined && (!Array.isArray(inventory.libraries) || inventory.libraries.length > 2000 || inventory.libraries.some(library => !library || typeof library !== 'object' || typeof library.id !== 'string' || !library.id || library.id.length > 200 || !['missing','downloading','verifying','ready','failed','outdated'].includes(library.state)))) throw new Error('Invalid game library inventory.'); return inventory; diff --git a/src/routes/hosts/[id]/+page.svelte b/src/routes/hosts/[id]/+page.svelte index 6a963f6..b33de6e 100644 --- a/src/routes/hosts/[id]/+page.svelte +++ b/src/routes/hosts/[id]/+page.svelte @@ -9,7 +9,7 @@ import { page } from '$app/stores'; import { goto } from '$app/navigation'; import type { Host, CapacityCalibration } from '$lib/hosts.js'; - import { hostAgentUpdateLabel, hostAgentUpdateMessage, hostAgentUpdatesEnabled, hostDeletionMessage, hostReadinessLabel } from '$lib/hosts.js'; + import { hostAgentLaneLabel, hostAgentLaneState, hostAgentUpdateLabel, hostAgentUpdateMessage, hostAgentUpdatesEnabled, hostDeletionMessage, hostReadinessLabel } from '$lib/hosts.js'; import SectionHeader from '$ui/components/primitives/SectionHeader.svelte'; import Button from '$ui/components/primitives/Button.svelte'; import DeleteHostDialog from '$ui/components/DeleteHostDialog.svelte'; @@ -85,6 +85,11 @@
{#if tab === 'overview'}
+

Agent workers

+ {#if host.inventory.agentLanes?.length}
{#each host.inventory.agentLanes as lane (lane.name)} +
{hostAgentLaneLabel(lane.name)}
{hostAgentLaneState(lane)}{#if !host.connected} (last reported){/if}
Last successful poll: {lane.lastPollAt ? new Date(lane.lastPollAt).toLocaleString() : 'Not yet reported'}
+ {/each}
{:else}

This agent has not reported worker status yet.

{/if} +

Inventory

Hostname
{host.hostname || 'Pending'}
Agent
{host.inventory.agentVersion ?? 'Unknown'}
Platform
{host.inventory.os ?? 'Unknown'} · {host.inventory.architecture ?? 'Unknown'}
CPU
{host.inventory.cpu ?? 'Unknown'} ({host.inventory.cpuCount ?? '?'} cores)
CPU usage
{host.inventory.cpuUsagePercent === undefined ? 'Not reported' : `${host.inventory.cpuUsagePercent.toFixed(1)}%`}{#if !host.connected && host.inventory.cpuUsagePercent !== undefined} (last reported){/if}
Memory available / total
{bytes(host.inventory.memoryAvailable)} / {bytes(host.inventory.memoryTotal)}
Game storage available / total
{bytes(host.inventory.storageAvailable)} / {bytes(host.inventory.storageTotal)}
Docker
{host.inventory.dockerVersion ?? 'Unknown'} · {host.inventory.dockerRunning ? 'Running' : 'Unavailable'}
Current operation
{host.inventory.currentOperation ?? 'None reported'}
{#if calibration}

Learned server capacity

diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index c48f6ae..22f2a18 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.5.2" +VERSION = "1.5.3" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -41,6 +41,7 @@ MANIFEST_MAX_BYTES = 16 * 1024 * 1024 MANIFEST_MAX_SECONDS = 15 MANIFEST_MAX_DEPTH = 64 LIBRARY_DOWNLOAD_WORKERS = 4 +HEALTH_LOCK_TIMEOUT_SECONDS = 5 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") AGENT_CONFIG = Path("/etc/electric-cherry/agent.json") AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py") @@ -923,6 +924,28 @@ class Agent: self.inventory_manifest_lock = threading.Lock() self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.executing_id = None + self.lane_lock = threading.Lock() + self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False} + for name in ("health-1", "health-2", "read", "write")} + + def lane_started(self, name): + with self.lane_lock: + self.lanes[name]["thread"] = threading.current_thread() + + def lane_polled(self, name): + with self.lane_lock: + self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic()) + + def lane_busy(self, name, busy): + with self.lane_lock: + self.lanes[name]["busy"] = busy + + def lane_inventory(self): + with self.lane_lock: + return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(), + "busy": lane["busy"], "lastPollAt": lane["lastPollAt"], + "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])} + for name, lane in self.lanes.items()] def can_uninstall(self): try: @@ -1334,7 +1357,7 @@ class Agent: os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest"] + (["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, **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)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} + 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, **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)), "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): health = self.inspect(instance_id, generation) @@ -2366,13 +2389,29 @@ class Agent: key = operation.get("instanceId") return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext() + @contextlib.contextmanager + def observation_lock(self, operation, health=False): + lock = self.instance_lock(operation) + if not health or not operation.get("instanceId"): + with lock: + yield + return + if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS): + raise AgentError("Health check timed out waiting for the instance operation") + try: + yield + finally: + lock.release() + @staticmethod def health_operation(operation): return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version")) - def read_loop(self, lane="read"): + def read_loop(self, lane="read", lane_name=None): # Each lane owns its SQLite connection and never writes executing_id, # checkpoints, or current_operation used by the mutation state machine. + lane_name = lane_name or ("health-1" if lane == "health" else "read") + self.lane_started(lane_name) journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False) next_prune = time.monotonic() try: @@ -2388,7 +2427,9 @@ class Agent: if journal.prune_observations(self.observation_keep_days) == 500: next_prune = time.monotonic() + 15 response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20") + self.lane_polled(lane_name) handled = False + self.lane_busy(lane_name, bool(response.get("operations"))) for operation in response.get("operations", []): if self.stopping.is_set() or not self.read_operation(operation): continue @@ -2405,12 +2446,12 @@ class Agent: row, _ = journal.accept(operation) if row["state"] in ("accepted", "unknown", "running"): self.report(journal.update(operation["id"], "running"), journal=journal) - with self.read_gate.read(), self.instance_lock(operation): - try: + try: + with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"): result = self.dispatch(operation) row = journal.update(operation["id"], "completed", result) - except Exception as err: - row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") + except Exception as err: + row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") # Receipt delivery can take seconds; it must not # hold the instance lock needed by health checks. self.report(row, journal=journal) @@ -2427,6 +2468,8 @@ class Agent: except Exception as err: print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr) self.stopping.wait(3) + finally: + self.lane_busy(lane_name, False) finally: journal.close() @@ -2440,6 +2483,8 @@ class Agent: return # Never hold the updater lock while waiting for delivery. response = self.request("GET", "/api/agent/operations?lane=write&wait=20") + self.lane_polled("write") + self.lane_busy("write", bool(response.get("operations"))) with self.activity() as acquired: if not acquired or self.stopping.is_set() or self.update_pending(): return @@ -2467,6 +2512,7 @@ class Agent: return completed def run(self): + self.lane_started("write") # Obtain the configured host port range before accepting any deployment. # Enrollment's default values must never race the first heartbeat. while not self.stopping.is_set(): @@ -2487,7 +2533,8 @@ class Agent: heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True) heartbeats.start() # Preserve two concurrent health checks even while files are being read. - readers = [threading.Thread(target=self.read_loop, args=(lane,), daemon=True) for lane in ("health", "health", "read")] + readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True) + for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))] for reader in readers: reader.start() while not self.stopping.is_set(): @@ -2499,6 +2546,8 @@ class Agent: print("Host controller contact unavailable; existing containers continue running", file=sys.stderr) except Exception as err: print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr) + finally: + self.lane_busy("write", False) self.stopping.wait(0.25 if completed else 3) for reader in readers: reader.join(timeout=2) diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 61e63d2..d9a21b3 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.5.2", + "version": "1.5.3", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "6e6a10a96f2392bf7595665bdff5b1b086d3446d5c43bd5770955aeebce9aa1d", + "sha256": "ebd6f67a44fdb4bd0683597d4abd54999311473561204de368864b04a5436256", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/host-agent-lanes.test.ts b/tests/unit/host-agent-lanes.test.ts new file mode 100644 index 0000000..4076257 --- /dev/null +++ b/tests/unit/host-agent-lanes.test.ts @@ -0,0 +1,45 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest'; +import { hostAgentLaneState, type HostAgentLane } from '../../src/lib/hosts.js'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +let fixture: Fixture; +const inventory = { protocolVersion:1, dockerRunning:true, directoriesWritable:true, storageAvailable:100 * 1024 ** 3, + runtimeImages:['image'], libraries:[{id:'library',state:'ready'}] }; +const lane: HostAgentLane = {name:'health-1',alive:true,busy:false,lastPollAt:'2026-10-06T12:00:00Z',pollAgeSeconds:5}; +beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); +afterEach(async () => { await fixture.cleanup(); }); + +it('flags a stopped or stalled health worker despite a live heartbeat and recovers on fresh polls', async () => { + const api = await import('../../src/lib/server/hosts/index.js'); + const host = api.createHost({name:'Worker host'}); + api.enrollHost(api.createEnrollment(host.id).token,inventory); + for (const broken of [{...lane,alive:false},{...lane,busy:true,pollAgeSeconds:91}]) { + const updated = api.heartbeat(host.id,{...inventory,agentLanes:[broken,{...lane,name:'health-2'}]}); + expect(updated).toMatchObject({connected:true,connectionState:'online',readiness:'degraded'}); + expect(updated.readinessReasons[0]).toContain('health check 1 worker'); + // Diagnostics must not prevent health polls from recovering the worker. + const check = api.queueHostOperation(host.id,'instance.inspect',{}); + expect(api.pendingHostOperations(host.id,'health').map(op=>op.id)).toContain(check.id); + } + expect(api.heartbeat(host.id,{...inventory,agentLanes:[lane]}).readiness).toBe('ready'); + // Older agents remain compatible without diagnostics. + expect(api.heartbeat(host.id,inventory).readiness).toBe('ready'); +}); + +it('allows long reads and mutations while detecting stalled idle workers and missing initial polls', () => { + expect(hostAgentLaneState({...lane,lastPollAt:null})).toBe('starting'); + expect(hostAgentLaneState({...lane,lastPollAt:null,pollAgeSeconds:91})).toBe('stalled'); + expect(hostAgentLaneState({...lane,busy:true,pollAgeSeconds:91})).toBe('stalled'); + for (const name of ['read','write'] as const) { + expect(hostAgentLaneState({...lane,name,busy:true,pollAgeSeconds:3600})).toBe('busy'); + expect(hostAgentLaneState({...lane,name,busy:false,pollAgeSeconds:91})).toBe('stalled'); + expect(hostAgentLaneState({...lane,name,alive:false,busy:true})).toBe('stopped'); + } +}); + +it('sanitizes lane diagnostics without rejecting a heartbeat for one malformed sample', async () => { + const { validateInventory } = await import('../../src/lib/server/hosts/index.js'); + expect(validateInventory({...inventory,agentLanes:[null,{...lane,secret:'omit'},{...lane},{...lane,name:'write',pollAgeSeconds:-1}]}).agentLanes) + .toEqual([lane]); + expect(validateInventory({...inventory,agentLanes:[{...lane,pollAgeSeconds:'stale'},{...lane,lastPollAt:'invalid'}]}).agentLanes).toEqual([]); +});