diff --git a/.env.example b/.env.example index 80f811a..013a54d 100644 --- a/.env.example +++ b/.env.example @@ -31,3 +31,6 @@ PLUTAINER_IMAGE=ghcr.io/ayymoss/plutainer:latest # Existing legacy registrations can still use their previous KOMODO_*, # SERVERS_ROOT, T6_SERVERS_ROOT, and RCON_HOST settings. Native hosts need none. # Fleet concurrency and campaign retention are saved in Settings, not env vars. + +# Optional request/report timings and event-loop delay diagnostics. +EC_PERF_LOG=false diff --git a/README.md b/README.md index 4ab4c82..fb3b067 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,20 @@ npm run test:agent npm run build ``` -Builds regenerate the downloadable agent, standalone updater and checksum manifest. Local controller +Builds regenerate the downloadable agent, standalone updater and checksum manifest, +and bundle the metrics/leaderboard workers under `build/reports`. `npm run dev` +and `npm test` compile those workers automatically. After editing a report +implementation during development, restart `npm run dev` to rebuild its worker. +Reports use separate read-only SQLite connections and bounded revision-aware caches; +metrics snapshots can be reused for up to 30 seconds when their inputs are unchanged. + +Set `EC_PERF_LOG=true` to log slow request timings, report computation/cache timings, +and event-loop delay. Responses include `Server-Timing`, and browser navigations +record an `ec:navigation` performance measure. `npm run benchmark:leaderboard` +uses an isolated synthetic database; pass `-- --scenario all --baseline-ref ` +to compare full response parity and warm query timings with an earlier revision. + +Local controller metadata defaults to `./data/cherry.db`; use a separate test database when exercising enrollment. Production agent transport requires HTTPS. diff --git a/agent/ec_agent.py b/agent/ec_agent.py index e8b9018..1d03a68 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.3.0" +VERSION = "1.4.0" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 LIBRARY_DOWNLOAD_WORKERS = 4 @@ -329,10 +329,47 @@ def managed_write(root, value, content, uid=1000, gid=1000): os.unlink(temp_name, dir_fd=parent) +class ReadWriteGate: + """Let observations run together; global mutations wait for existing readers.""" + def __init__(self): + self.condition = threading.Condition() + self.readers = 0 + self.writers = 0 + + @contextlib.contextmanager + def read(self): + with self.condition: + while self.writers: + self.condition.wait() + self.readers += 1 + try: + yield + finally: + with self.condition: + self.readers -= 1 + self.condition.notify_all() + + @contextlib.contextmanager + def write(self): + # The mutation lane is serial, so there is only one writer. + with self.condition: + self.writers += 1 + while self.readers: + self.condition.wait() + try: + yield + finally: + with self.condition: + self.writers -= 1 + self.condition.notify_all() + + class Journal: - def __init__(self, path): + def __init__(self, path, recover=True): self.db = sqlite3.connect(path) self.db.row_factory = sqlite3.Row + if not recover: + return self.db.execute("PRAGMA journal_mode=WAL") self.db.execute("PRAGMA synchronous=FULL") self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)") @@ -672,6 +709,10 @@ class Agent: self.docker = docker or Docker() self.stopping = threading.Event() self.current_operation = None + self.instance_locks = {} + self.read_gate = ReadWriteGate() + self.read_claim_lock = threading.Lock() + self.read_claims = set() self.cancelled_creations = frozenset() self.library_progress = {} self.library_transfers = {} @@ -732,17 +773,25 @@ class Agent: return Path(self.config_path) == AGENT_CONFIG and (UPDATE_JOURNAL.exists() or UPDATE_JOURNAL.is_symlink()) @contextlib.contextmanager - def activity(self): + def activity(self, read_only=False): path = self.root / "agent" / "activity.lock" descriptor = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + mutation_descriptor = None try: try: - fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + # All work excludes the updater. Mutations additionally exclude + # other mutation batches while read lanes may continue. + fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB) + if not read_only: + mutation_descriptor = os.open(self.root / "agent" / "mutation.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + fcntl.flock(mutation_descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: yield False return yield True finally: + if mutation_descriptor is not None: + os.close(mutation_descriptor) os.close(descriptor) def record_health(self): @@ -953,7 +1002,7 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["instance.cancel-create"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) + capabilities = ["instance.cancel-create", "operations.read-lane"] + (["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): @@ -1860,23 +1909,89 @@ class Agent: except (AgentError, OSError, ValueError): pass + @staticmethod + def read_operation(operation): + return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version")) + + def instance_lock(self, operation): + key = operation.get("instanceId") + return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext() + + def read_loop(self): + # Each lane owns its SQLite connection and never writes executing_id, + # checkpoints, or current_operation used by the mutation state machine. + journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False) + try: + while not self.stopping.is_set(): + try: + if self.update_pending(): + self.stopping.wait(3) + continue + response = self.request("GET", "/api/agent/operations?lane=read&wait=20") + for operation in response.get("operations", []): + if self.stopping.is_set() or not self.read_operation(operation): + continue + with self.read_claim_lock: + if operation["id"] in self.read_claims: + continue + self.read_claims.add(operation["id"]) + try: + with self.activity(read_only=True) as acquired: + if not acquired or self.update_pending(): + continue + with self.read_gate.read(), self.instance_lock(operation): + row, _ = journal.accept(operation) + if row["state"] in ("accepted", "unknown", "running"): + self.report(journal.update(operation["id"], "running")) + try: + 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") + self.report(row) + finally: + with self.read_claim_lock: + self.read_claims.discard(operation["id"]) + except (AgentError, OSError, ValueError, KeyError, sqlite3.Error): + self.stopping.wait(3) + finally: + journal.close() + def poll_operations(self): + if self.stopping.is_set() or self.update_pending(): + return + # Check both update and mutation locks before delivery, release them + # during the long poll, and recheck immediately before accepting work. + with self.activity() as acquired: + if not acquired: + return + # Never hold the updater lock while waiting for delivery. + response = self.request("GET", "/api/agent/operations?lane=write&wait=20") with self.activity() as acquired: if not acquired or self.stopping.is_set() or self.update_pending(): return if self.config.get("pendingCredential"): self.recover_credential() - response = self.request("GET", "/api/agent/operations") + completed = True for operation in response.get("operations", []): if self.stopping.is_set(): break + if self.read_operation(operation): + continue self.current_operation = operation["id"] + # Global changes wait for observations; instance changes only + # block observations for their own immutable instance identity. + guard = self.read_gate.write() if operation["kind"] in ("host.decommission", "credentials.rotate") else contextlib.nullcontext() try: - accepted, _ = self.journal.accept(operation) - self.report(accepted) - self.report(self.execute(operation, report_running=True)) + with guard, self.instance_lock(operation): + accepted, _ = self.journal.accept(operation) + self.report(accepted) + result = self.execute(operation, report_running=True) + self.report(result) + completed = completed and result["state"] in ("completed", "failed") finally: self.current_operation = None + return completed def run(self): # Obtain the configured host port range before accepting any deployment. @@ -1895,13 +2010,19 @@ class Agent: self.stopping.wait(3) heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True) heartbeats.start() + readers = [threading.Thread(target=self.read_loop, daemon=True) for _ in range(2)] + for reader in readers: + reader.start() while not self.stopping.is_set(): + completed = False try: - self.poll_operations() + completed = self.poll_operations() except (AgentError, OSError, ValueError, KeyError): # Logs deliberately omit URLs, tokens, payloads and raw errors. print("Host controller contact unavailable; existing containers continue running", file=sys.stderr) - self.stopping.wait(3) + self.stopping.wait(0.25 if completed else 3) + for reader in readers: + reader.join(timeout=2) heartbeats.join(timeout=2) diff --git a/agent/tests/test_update_integration.py b/agent/tests/test_update_integration.py index d4ee585..4228ce9 100644 --- a/agent/tests/test_update_integration.py +++ b/agent/tests/test_update_integration.py @@ -4,6 +4,7 @@ import json import os from pathlib import Path import tempfile +import threading import unittest from unittest.mock import patch @@ -43,11 +44,15 @@ class UpdateIntegrationTests(unittest.TestCase): with self.assertRaises(BlockingIOError): fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) - def test_operation_delivery_execution_and_reports_exclude_updates(self): - operation = {"id": "one", "kind": "host.inspect", "payload": {}} + def test_execution_and_reports_exclude_updates_but_long_poll_does_not(self): + operation = {"id": "one", "kind": "runtime.pull", "payload": {}} calls = [] def request(method, path, body=None): - self.assert_activity_held() + if method != "GET": + self.assert_activity_held() + else: + with (self.agent.root / "agent" / "activity.lock").open("a") as stream: + fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) calls.append((method, path)) return {"operations": [operation]} if method == "GET" else {} def dispatch(value): @@ -82,7 +87,61 @@ class UpdateIntegrationTests(unittest.TestCase): module.UPDATE_JOURNAL.unlink() with patch.object(self.agent, "request", return_value={"operations": []}) as request: self.agent.poll_operations() - request.assert_called_once_with("GET", "/api/agent/operations") + request.assert_called_once_with("GET", "/api/agent/operations?lane=write&wait=20") + + def test_read_lane_completes_while_mutation_is_busy_and_excludes_updater(self): + read_done = threading.Event() + operation = {"id": "read-one", "kind": "host.inspect", "payload": {}} + self.agent.request = lambda method, path, body=None: {"operations": [operation]} if method == "GET" else {} + def dispatch(value): + self.assert_activity_held() + read_done.set() + self.agent.stopping.set() + return {"ok": True} + self.agent.dispatch = dispatch + with self.agent.activity() as acquired: + self.assertTrue(acquired) + reader = threading.Thread(target=self.agent.read_loop) + reader.start() + self.assertTrue(read_done.wait(2), "Read waited behind unrelated mutation activity") + reader.join(2) + self.assertFalse(reader.is_alive()) + self.assertEqual(self.agent.journal.get("read-one")["state"], "completed") + self.assertIsNone(self.agent.executing_id) + + def test_read_lane_never_dispatches_mutations(self): + def request(method, path, body=None): + self.agent.stopping.set() + return {"operations": [{"id": "bad-read", "kind": "instance.remove", "payload": {}}]} + self.agent.request = request + with patch.object(self.agent, "dispatch") as dispatch: + self.agent.read_loop() + dispatch.assert_not_called() + + def test_instance_mutations_and_global_mutations_exclude_observations(self): + operation = {"instanceId": "one"} + entered = threading.Event() + release = threading.Event() + def read(): + with self.agent.read_gate.read(), self.agent.instance_lock(operation): + entered.set() + release.wait(2) + with self.agent.instance_lock(operation): + reader = threading.Thread(target=read) + reader.start() + self.assertFalse(entered.wait(0.05)) + self.assertTrue(entered.wait(2)) + writer_entered = threading.Event() + def write(): + with self.agent.read_gate.write(): + writer_entered.set() + writer = threading.Thread(target=write) + writer.start() + self.assertFalse(writer_entered.wait(0.05)) + release.set() + reader.join(2) + writer.join(2) + self.assertTrue(writer_entered.is_set()) def test_failed_controller_contact_never_writes_healthy_receipt(self): with patch.object(self.agent, "request", side_effect=module.AgentError("offline")): diff --git a/docs/native-hosts.md b/docs/native-hosts.md index a2fe461..fc43577 100644 --- a/docs/native-hosts.md +++ b/docs/native-hosts.md @@ -443,3 +443,23 @@ check can install the release once the batch has drained. When a heartbeat advertises cancellation support, delivery resumes with cancellation cleanup first. This controller-side hold also recovers cancellations already waiting on older agents. Hosts without automatic updates need the current installer. + +## Responsive observations (agent 1.4.0) + +Agent 1.4.0 separates read-only inspections, file/log reads, and the exact RCON +commands `status`, `getstatus`, and `version` from the serial mutation queue. +Two bounded read lanes can run during runtime or library downloads. Instance +mutations exclude reads of that instance; credential rotation and host cleanup +wait for active reads. Both lanes retain durable operation IDs, generation checks, +and the updater activity lock. Older agents continue using the combined queue. + +Agents long-poll `GET /api/agent/operations?lane=read|write&wait=20`; requests +without those parameters retain immediate combined delivery. The controller wakes +pending polls when work is queued and rechecks authentication after waiting. +Long polls release the updater lock; execution reacquires it and rechecks update +state before accepting work. + +Browser game-folder uploads transfer up to three files concurrently. Chunks of +one file remain serial and checksum-verified. Publication waits for admitted file +transfers and prevents subsequent writes; pause/resume still uses acknowledged +per-file offsets. Host downloads retain their existing bounded concurrency. diff --git a/package.json b/package.json index 48a8e97..863578d 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "type": "module", "scripts": { "dev": "vite dev", - "build": "vite build", + "build": "vite build && npm run build:reports -- --outDir build/reports", "preview": "vite preview", "check": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json", "test": "vitest run", @@ -15,7 +15,11 @@ "seed:fixture": "tsx scripts/seed-fixture.ts", "agent:package": "python3 agent/package_release.py", "prebuild": "npm run agent:package", - "test:agent": "python3 -m unittest discover -s agent/tests -v" + "test:agent": "python3 -m unittest discover -s agent/tests -v", + "build:reports": "vite build --config vite.reports.config.ts", + "predev": "npm run build:reports", + "pretest": "npm run build:reports", + "pretest:watch": "npm run build:reports" }, "dependencies": { "@fontsource/alfa-slab-one": "^5.3.0", diff --git a/scripts/benchmark-leaderboard.mjs b/scripts/benchmark-leaderboard.mjs index 3ff24ed..becdc1c 100644 --- a/scripts/benchmark-leaderboard.mjs +++ b/scripts/benchmark-leaderboard.mjs @@ -45,6 +45,7 @@ const percentile = (values, fraction) => [...values].sort((a, b) => a - b)[Math. function seed(handle, definition) { handle.exec(source('src/lib/server/db/schema.sql')); + handle.exec("CREATE TABLE report_revisions(topic TEXT PRIMARY KEY, revision INTEGER NOT NULL); INSERT INTO report_revisions VALUES ('leaderboard',0)"); const match = handle.prepare(`INSERT INTO leaderboard_matches (id, server, game, map, map_key, map_name, started_at, ended_at, last_received_at, status, start_observed, start_players, highest_round, last_event, last_fingerprint) diff --git a/src/hooks.server.ts b/src/hooks.server.ts index 25b4c22..8f5e151 100644 --- a/src/hooks.server.ts +++ b/src/hooks.server.ts @@ -3,6 +3,8 @@ import { migrate } from '$lib/server/db/migrate.js'; import { startWorker } from '$lib/server/pipeline/worker.js'; import { interruptDeployments } from '$lib/server/provisioning/history.js'; import { authenticationHandle } from '$lib/server/auth/guard.js'; +import { emit } from '$lib/server/pipeline/bus.js'; +import { startPerformanceMonitor } from '$lib/server/performance.js'; let booted = false; function boot(): void { @@ -11,6 +13,7 @@ function boot(): void { migrate(); interruptDeployments(); startWorker(); + startPerformanceMonitor(); } // SvelteKit calls handle() for every request; using the first request as the @@ -18,5 +21,13 @@ function boot(): void { // out-of-band init hook. export const handle: Handle = async (input) => { boot(); - return authenticationHandle(input); + const started = performance.now(); + const response = await authenticationHandle(input); + const elapsed = performance.now() - started; + response.headers.append('Server-Timing', `app;dur=${elapsed.toFixed(1)}`); + if (process.env.EC_PERF_LOG === 'true' && elapsed > 250 && !input.event.route.id?.startsWith('/api/agent/')) { + console.info('[request]', { route: input.event.route.id, method: input.event.request.method, status: response.status, elapsedMs: Math.round(elapsed) }); + } + if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) emit({ type: 'fleet-invalidated' }); + return response; }; diff --git a/src/lib/server/db/index.ts b/src/lib/server/db/index.ts index 1b3cf4a..8d8dfc3 100644 --- a/src/lib/server/db/index.ts +++ b/src/lib/server/db/index.ts @@ -1,4 +1,5 @@ import Database from 'better-sqlite3'; +import { installReportRevisions } from '../reports/revisions.js'; import { mkdirSync } from 'node:fs'; import { dirname, resolve } from 'node:path'; import { config } from '../config.js'; @@ -368,6 +369,7 @@ export function db(): Database.Database { backfillLeaderboard(handle); initializeFleetSettings(handle); migrateNativeInstances(handle); + installReportRevisions(handle); } catch (err) { handle.close(); throw err; diff --git a/src/lib/server/fleet/leaderboard.ts b/src/lib/server/fleet/leaderboard.ts index da4d90a..326fd51 100644 --- a/src/lib/server/fleet/leaderboard.ts +++ b/src/lib/server/fleet/leaderboard.ts @@ -10,6 +10,11 @@ const metrics = new Set(leaderboardMetrics.map(metric => metric.key)); export const MAX_LEADERBOARD_OFFSET = 1_000_000; type RankedEntry = Omit; +interface RankedScope { revision: number; entries: RankedEntry[]; maps: LeaderboardMap[]; summary: LeaderboardResponse['summary'] } +const scopeCaches = new WeakMap>(); +// Match SQLite's built-in lower(), which folds ASCII rather than Unicode. +const searchText = (value: string) => value.replace(/[A-Z]/g, ch => ch.toLowerCase()); + interface LeaderboardRow extends Omit { // A left join keeps metadata available when the requested page has no players. guid: string | null; @@ -91,21 +96,15 @@ function readLeaderboard(options: LeaderboardOptions, exactGuid?: string): Leade SELECT measures.*, p.name, RANK() OVER (ORDER BY measures.${metric} DESC) AS rank FROM measures JOIN identities p ON p.guid = measures.guid AND p.recency = 1 ${qualification} )`; - if (exactGuid !== undefined) { - const entries = handle.prepare(`${aggregated} SELECT * FROM ranked WHERE guid = ?`) - .all(...parameters, exactGuid) as RankedEntry[]; - return addRecordSources(handle, entries, options)[0] ?? null; - } - - const query = options.q?.trim() ?? ''; - const search = query ? 'WHERE instr(lower(name), lower(?)) > 0 OR instr(lower(guid), lower(?)) > 0' : ''; - // Reuse one ranked set for the page and matching count. Summaries deliberately - // cover the full game/map scope, including players below an efficiency threshold. - const rows = handle.prepare(`${aggregated}, searched AS MATERIALIZED ( - SELECT * FROM ranked ${search} - ), page AS ( - SELECT * FROM searched ORDER BY ${metric} DESC, name COLLATE NOCASE, name, guid LIMIT ? OFFSET ? - ), playerSummary AS ( + let revision = -1; + try { revision = (handle.prepare("SELECT revision FROM report_revisions WHERE topic='leaderboard'").get() as { revision: number }).revision; } + catch { /* Standalone history tools can operate without cache metadata. */ } + let cache = scopeCaches.get(handle); + if (!cache) { cache = new Map(); scopeCaches.set(handle, cache); } + const cacheKey = JSON.stringify([options.game ?? '', options.map ?? '', metric]); + let saved = cache.get(cacheKey); + if (!saved || revision < 0 || saved.revision !== revision) { + const rows = handle.prepare(`${aggregated}, playerSummary AS ( SELECT COUNT(*) AS players, COALESCE(SUM(kills), 0) AS kills, COALESCE(SUM(roundsSurvived), 0) AS roundsSurvived, COALESCE(MAX(highestRound), 0) AS highestRound FROM totals @@ -113,21 +112,18 @@ function readLeaderboard(options: LeaderboardOptions, exactGuid?: string): Leade SELECT COUNT(DISTINCT match_id) AS matches, MIN(matchStartedAt) AS firstTrackedAt, MAX(matchUpdatedAt) AS updatedAt FROM observations ), metadata AS ( - SELECT (SELECT COUNT(*) FROM searched) AS total, playerSummary.*, matchSummary.* - FROM playerSummary CROSS JOIN matchSummary - ) SELECT page.*, metadata.total AS matchingTotal, metadata.players AS summaryPlayers, + SELECT playerSummary.*, matchSummary.* FROM playerSummary CROSS JOIN matchSummary + ) SELECT ranked.*, metadata.players AS summaryPlayers, metadata.matches AS summaryMatches, metadata.kills AS summaryKills, metadata.roundsSurvived AS summaryRounds, metadata.highestRound AS summaryHighestRound, metadata.firstTrackedAt AS summaryFirstTrackedAt, metadata.updatedAt AS summaryUpdatedAt - FROM metadata LEFT JOIN page ON 1 - ORDER BY page.${metric} DESC, page.name COLLATE NOCASE, page.name, page.guid`) - .all(...parameters, ...(query ? [query, query] : []), limit, offset) as LeaderboardRow[]; + FROM metadata LEFT JOIN ranked ON 1 + ORDER BY ranked.${metric} DESC, ranked.name COLLATE NOCASE, ranked.name, ranked.guid`) + .all(...parameters) as LeaderboardRow[]; const rankedEntries = rows.filter((row): row is LeaderboardRow & { guid: string } => row.guid !== null) .map(({ matchingTotal, summaryPlayers, summaryMatches, summaryKills, summaryRounds, summaryHighestRound, summaryFirstTrackedAt, summaryUpdatedAt, ...entry }) => entry); - const entries = addRecordSources(handle, rankedEntries, options); const metadata = rows[0]; - const total = metadata.matchingTotal; const summary: LeaderboardResponse['summary'] = { players: metadata.summaryPlayers, matches: metadata.summaryMatches, kills: metadata.summaryKills, roundsSurvived: metadata.summaryRounds, highestRound: metadata.summaryHighestRound, @@ -140,5 +136,18 @@ function readLeaderboard(options: LeaderboardOptions, exactGuid?: string): Leade GROUP BY game, map_key HAVING id = MAX(id) ORDER BY game, name COLLATE NOCASE, map_key`) .all(...(options.game ? [options.game] : [])) as LeaderboardMap[]; - return { entries, maps, total, limit, offset, metric, summary }; + saved = { revision, entries: rankedEntries, maps, summary }; + cache.delete(cacheKey); + if (rankedEntries.length <= 100_000) cache.set(cacheKey, saved); + // Bound total retained players as well as the number of scopes. + while (cache.size > 8 || [...cache.values()].reduce((n, scope) => n + scope.entries.length, 0) > 100_000) cache.delete(cache.keys().next().value!); + } + if (exactGuid !== undefined) { + const entry = saved.entries.find(entry => entry.guid === exactGuid); + return entry ? addRecordSources(handle, [entry], options)[0] : null; + } + const query = searchText(options.q?.trim() ?? ''); + const matching = query ? saved.entries.filter(entry => searchText(entry.name).includes(query) || searchText(entry.guid).includes(query)) : saved.entries; + const entries = addRecordSources(handle, matching.slice(offset, offset + limit), options); + return { entries, maps: saved.maps, total: matching.length, limit, offset, metric, summary: saved.summary }; } diff --git a/src/lib/server/fleet/servers.ts b/src/lib/server/fleet/servers.ts index ffb51ce..8d7ba29 100644 --- a/src/lib/server/fleet/servers.ts +++ b/src/lib/server/fleet/servers.ts @@ -179,6 +179,10 @@ export function setServerField( field: keyof ServerRow, value: string | number | null ): void { + setServerFields(name, { [field]: value }); +} + +export function setServerFields(name: string, fields: Partial>): void { const allowed: Array = [ 'komodo_type', 'komodo_name', @@ -211,8 +215,12 @@ export function setServerField( 'game_ended_at', 'game_cycle_key' ]; - if (!allowed.includes(field)) throw new Error(`refusing to set ${field}`); - db().prepare(`UPDATE servers SET ${field} = ? WHERE name = ?`).run(value, name); + const entries = Object.entries(fields); + if (!entries.length) return; + for (const [field] of entries) if (!allowed.includes(field as keyof ServerRow)) throw new Error(`refusing to set ${field}`); + db().prepare(`UPDATE servers SET ${entries.map(([field]) => `${field} = ?`).join(', ')} WHERE name = ? + AND (${entries.map(([field]) => `${field} IS NOT ?`).join(' OR ')})`) + .run(...entries.map(([,value]) => value), name, ...entries.map(([,value]) => value)); } /** diff --git a/src/lib/server/fleet/snapshot.ts b/src/lib/server/fleet/snapshot.ts new file mode 100644 index 0000000..f92ae74 --- /dev/null +++ b/src/lib/server/fleet/snapshot.ts @@ -0,0 +1,51 @@ +import { db } from '../db/index.js'; +import { listServers } from './servers.js'; +import { observationReader } from './status.js'; +import { config, komodoConfigured, rconConfigured } from '../config.js'; +import { runningVersions } from '../integrations/plutonium.js'; +import { listHosts } from '../hosts/index.js'; +import type { NativeInstance } from '../backends/native.js'; + +export function fleetSnapshot() { + const rconHost = config.rcon.host; + const hosts = listHosts(); + const hostsById = new Map(hosts.map(({ inventory: _inventory, ...host }) => [host.id, host])); + const versions = new Map(runningVersions().map(version => [version.server, version])); + const instances = new Map((db().prepare('SELECT * FROM native_instances WHERE retired_at IS NULL').all() as NativeInstance[]).map(instance => [instance.server, instance])); + const holds = new Map(); + for (const row of db().prepare(`SELECT bsv.server, s.target_rel_path AS target, b.id AS betaId + FROM beta_servers bsv JOIN betas b ON b.id=bsv.beta_id JOIN beta_scripts bs ON bs.beta_id=b.id + JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')`).all() as + { server: string; target: string; betaId: number }[]) { + const entries = holds.get(row.server) ?? []; entries.push({ target: row.target, betaId: row.betaId }); holds.set(row.server, entries); + } + const observe = observationReader(); + const servers = listServers().map((s) => { + const instance = s.native_instance_id ? instances.get(s.name) : null; + return { + ...observe(s), + host: s.host_id ? hostsById.get(s.host_id) ?? null : null, + network: instance ? { bindAddress: instance.bind_address, localPort: instance.port, publicAddress: instance.public_address, publicPort: instance.public_port } : null, + plutonium: versions.get(s.name), + // Derived endpoint — same logic as endpointFor() but without asking + // Komodo. If we don't have a port yet, endpoint is null. + endpoint: instance ? { host: `agent:${instance.host_id}`, port: instance.port } : s.game_port ? { host: s.rcon_host ?? rconHost, port: s.game_port } : null, + held: holds.get(s.name) ?? [] + }; + }); + return { + servers, + serverNow: Date.now(), + reservedNames: (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server), + integrations: { + native: hosts.length > 0 || !komodoConfigured(), + hostsReady: hosts.filter(host => host.readiness === 'ready').length, + hostsTotal: hosts.length, + rcon: rconConfigured(), + komodo: komodoConfigured(), + rconTimeoutMs: config.rcon.timeoutMs, + defaultRconHost: rconHost + } + }; +} + diff --git a/src/lib/server/fleet/status.ts b/src/lib/server/fleet/status.ts index ba129c1..9650304 100644 --- a/src/lib/server/fleet/status.ts +++ b/src/lib/server/fleet/status.ts @@ -4,21 +4,57 @@ import { isProvisioning } from './provisioning.js'; import { restartSettling } from '../provisioning/restarts.js'; import { type ServerRow } from './servers.js'; import { telemetryReport, latestGameSubmission } from './status-evidence.js'; +import { db } from '../db/index.js'; +import { parseGameEvent, type GameTelemetry } from '../../telemetry.js'; +import { restartRequestInFlight } from '../provisioning/restarts.js'; + +interface Evidence { provisioning: boolean; settling: boolean; telemetry: GameTelemetry | null; submission: string | null } + +/** One set of evidence reads per snapshot, rather than four queries per server. */ +export function observationReader() { + const handle = db(); + const operations = new Map((handle.prepare(`SELECT server, action, state FROM server_deployments + WHERE rowid IN (SELECT MAX(rowid) FROM server_deployments WHERE server_deleted_at IS NULL GROUP BY server)`).all() as + { server: string; action: string; state: string }[]).map(row => [row.server, row])); + const restarts = new Map((handle.prepare('SELECT server, requested_at, healthy_at FROM restart_rollout').all() as + { server: string; requested_at: string; healthy_at: string | null }[]).map(row => [row.server, row])); + const submissions = new Map((handle.prepare(`SELECT server, MAX(at) AS at FROM ( + SELECT server, submitted_at AS at FROM server_deployments WHERE server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild') + UNION ALL SELECT server, requested_at AS at FROM restart_rollout) GROUP BY server`).all() as + { server: string; at: string | null }[]).map(row => [row.server, row.at])); + const reports = new Map(); + for (const row of handle.prepare('SELECT server, received_at, payload FROM server_telemetry WHERE payload IS NOT NULL').all() as + { server: string; received_at: string; payload: string }[]) { + try { + const saved = JSON.parse(row.payload) as GameTelemetry; + const event = parseGameEvent(saved); + if (event.server === row.server) reports.set(row.server, { ...event, receivedAt: row.received_at, timing: saved.timing }); + } catch { /* Invalid retained telemetry is unavailable. */ } + } + return (server: ServerRow) => { + const operation = operations.get(server.name), restart = restarts.get(server.name); + return serverObservation(server, { + provisioning: !!operation && ['create','rebuild','start'].includes(operation.action) && ['queued','running','awaiting-health'].includes(operation.state), + settling: restartRequestInFlight(server.name) || !!(restart && !restart.healthy_at && Date.now() - Date.parse(restart.requested_at) < 15_000), + telemetry: reports.get(server.name) ?? null, submission: submissions.get(server.name) ?? null + }); + }; +} /** Shared snapshot for REST, status events, metrics, and other server-side readers. */ -export function serverObservation(server: ServerRow) { +export function serverObservation(server: ServerRow, evidence?: Evidence) { let roster = null; try { roster = server.last_roster ? JSON.parse(server.last_roster) : null; } catch { /* unavailable */ } const rcon: GameObservation = { last_players: server.last_players, last_map: server.last_map, last_gametype: server.last_gametype, last_seen_at: server.last_seen_at, last_error: server.last_error, roster }; - const provisioning = isProvisioning(server.name); - const state = restartSettling(server.name) ? 'restarting' : server.container_state; + const provisioning = evidence ? evidence.provisioning : isProvisioning(server.name); + const state = (evidence ? evidence.settling : restartSettling(server.name)) ? 'restarting' : server.container_state; const ready = !provisioning && (!state || state === 'running'); const health = provisioning ? 'provisioning' : classifyHealth(server.last_seen_at, server.last_error, state); return selectServerObservation({ ...server, ...rcon, rcon_status: rcon, container_state: state, health, status_source: null as 'telemetry' | 'rcon' | null, - telemetry: telemetryReport(server.name) ?? undefined, - telemetry_after: latestGameSubmission(server.name), + telemetry: (evidence ? evidence.telemetry : telemetryReport(server.name)) ?? undefined, + telemetry_after: evidence ? evidence.submission : latestGameSubmission(server.name), last_error: ready ? server.last_error : null, replySnippet: ready ? server.last_reply_snippet : null, active_since: ready ? server.active_since : null, inactive_since: ready ? server.inactive_since : null, diff --git a/src/lib/server/game-files/index.ts b/src/lib/server/game-files/index.ts index e3a7cdb..c2e9576 100644 --- a/src/lib/server/game-files/index.ts +++ b/src/lib/server/game-files/index.ts @@ -18,14 +18,14 @@ type CacheRow = { host_id: string; version_id: string; status: LibraryCacheState 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[]; +function versionFromRow(row: VersionRow & { file_count?: number }, summaryCaches?: Map): GameLibraryVersion { + const files = summaryCaches ? [] : JSON.parse(row.manifest) as GameFileEntry[]; return { id: row.id, libraryId: row.library_id, game: row.game, version: row.version, description: row.description, state: row.state, sizeBytes: row.size_bytes, - files, fileCount: files.length || (database().prepare('SELECT COUNT(*) AS count FROM game_file_upload_entries WHERE version_id=?').get(row.id) as { count: number }).count, manifestSha256: row.manifest_sha256, createdAt: row.created_at, + files, fileCount: row.file_count ?? (files.length || (database().prepare('SELECT COUNT(*) AS count FROM game_file_upload_entries WHERE version_id=?').get(row.id) as { count: number }).count), manifestSha256: row.manifest_sha256, createdAt: row.created_at, publishedAt: row.published_at, storageLocation: row.storage_location, sourceHostId: row.source_host_id || null, sourceVolume: row.source_volume || null, error: row.error, operationId: row.operation_id, - caches: (database().prepare('SELECT * FROM host_library_caches WHERE version_id = ?').all(row.id) as CacheRow[]).map(cacheFromRow) }; + caches: summaryCaches ? summaryCaches.get(row.id) ?? [] : (database().prepare('SELECT * FROM host_library_caches WHERE version_id = ?').all(row.id) as CacheRow[]).map(cacheFromRow) }; } export function getLibraryVersion(id: string): GameLibraryVersion | null { const row = database().prepare('SELECT v.*, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id WHERE v.id=?').get(id) as VersionRow | undefined; @@ -52,12 +52,19 @@ export function latestPublishedLibraries() { return version ? [{ ...version, name: game.name }] : []; }); } -export function listLibraries(): GameLibrary[] { +export function listLibraries(summary = false): GameLibrary[] { reconcileLibraryOperations(); + let summaryCaches: Map | undefined; + if (summary) { + summaryCaches = new Map(); + for (const row of database().prepare('SELECT * FROM host_library_caches').all() as CacheRow[]) { + const items = summaryCaches.get(row.version_id) ?? []; items.push(cacheFromRow(row)); summaryCaches.set(row.version_id, items); + } + } const latest = new Map(latestPublishedLibraries().map(version => [version.game, version.id])); return GAME_FILE_GAMES.map(game => { - const versions = (database().prepare(`SELECT v.*, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id - WHERE l.game=? ORDER BY v.created_at DESC,v.rowid DESC`).all(game.id) as VersionRow[]).map(versionFromRow); + const versions = (database().prepare(`SELECT ${summary ? `v.id, v.library_id, v.version, v.description, v.dlc, v.state, v.size_bytes, v.manifest_sha256, v.created_at, v.published_at, v.storage_location, v.source_host_id, v.source_volume, v.error, v.operation_id, CASE WHEN json_array_length(v.manifest)>0 THEN json_array_length(v.manifest) ELSE (SELECT COUNT(*) FROM game_file_upload_entries e WHERE e.version_id=v.id) END AS file_count` : 'v.*'}, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id + WHERE l.game=? ORDER BY v.created_at DESC,v.rowid DESC`).all(game.id) as VersionRow[]).map(row => versionFromRow(row, summaryCaches)); const firstLibrary = database().prepare('SELECT created_at FROM game_file_libraries WHERE game=? ORDER BY created_at,rowid LIMIT 1').get(game.id) as { created_at: string } | undefined; return { id: game.id, name: game.name, game: game.id, description: '', createdAt: firstLibrary?.created_at ?? '', latestVersionId: latest.get(game.id) ?? null, versions }; }); @@ -392,9 +399,9 @@ export function reconcileLibraryOperations(): void { /** Lightweight list used in launch forms; large manifests stay on the detail endpoint. */ export function libraryOverview() { const hosts = listHosts(); - return { libraries: listLibraries().map(library => ({ ...library, versions: library.versions.map(({ files: _files, ...version }) => ({ + return { libraries: listLibraries(true).map(library => ({ ...library, versions: library.versions.map(({ files: _files, ...version }) => ({ ...version, caches: hosts.map(host => { - const cached = getHostLibraryCache(host.id, version.id); + const cached = version.caches.find(cache => cache.hostId === host.id); if (cached) return { ...cached, status: cached.status === 'ready' && library.latestVersionId !== version.id ? 'outdated' as const : cached.status }; return { hostId: host.id, libraryVersionId: version.id, status: 'missing' as const, bytesDownloaded: 0, diff --git a/src/lib/server/game-files/uploads.ts b/src/lib/server/game-files/uploads.ts index fcbf595..8252ed5 100644 --- a/src/lib/server/game-files/uploads.ts +++ b/src/lib/server/game-files/uploads.ts @@ -44,20 +44,50 @@ export function validateUploadPlan(input: unknown): UploadFilePlan[] { return manifest.map(file => ({ path: file.path, size: file.size, lastModified: modified.get(file.path)! })); } -const uploadLocks = new Map>(); +interface UploadLocks { barrier: Promise; writers: number; files: Map>; active: Set> } +const uploadLocks = new Map(); +function locksFor(key: string): UploadLocks { + let locks = uploadLocks.get(key); + if (!locks) { locks = { barrier: Promise.resolve(), writers: 0, files: new Map(), active: new Set() }; uploadLocks.set(key, locks); } + return locks; +} +function releaseLocks(key: string, locks: UploadLocks) { + if (!locks.writers && !locks.active.size && uploadLocks.get(key) === locks) uploadLocks.delete(key); +} /** Serialize transfers and final publication for one import; a rejected operation never poisons the lock. */ export async function withUploadLock(root: string, callback: () => Promise | T): Promise { const key = resolve(root); - const previous = uploadLocks.get(key) ?? Promise.resolve(); + const locks = locksFor(key); + // Capture readers admitted before publication, then prevent later readers + // from starting until publication finishes (including failed publication). + const previous = Promise.all([locks.barrier, ...locks.active]); let release!: () => void; const current = new Promise(done => { release = done; }); - uploadLocks.set(key, current); + locks.barrier = current; + locks.writers++; await previous; try { return await callback(); } finally { release(); - if (uploadLocks.get(key) === current) uploadLocks.delete(key); + locks.writers--; + releaseLocks(key, locks); + } +} + +/** Different files may transfer together; offsets for the same file remain serial. */ +export async function withUploadFileLock(root: string, file: string, callback: () => Promise): Promise { + const key = resolve(root), locks = locksFor(key); + const previous = Promise.all([locks.barrier, locks.files.get(file)]); + let release!: () => void; + const current = new Promise(done => { release = done; }); + locks.files.set(file, current); locks.active.add(current); + await previous; + try { return await callback(); } + finally { + release(); locks.active.delete(current); + if (locks.files.get(file) === current) locks.files.delete(file); + releaseLocks(key, locks); } } @@ -95,7 +125,7 @@ export async function receiveUploadChunk( } if (!body && chunk.size !== 0) throw new Error('A file chunk body is required.'); - return withUploadLock(root, async () => { + return withUploadFileLock(root, file.path, async () => { await beforeWrite?.(); const progress = await getUploadFileProgress(root, file); if (chunk.offset !== progress.receivedBytes) throw new UploadOffsetError(progress.receivedBytes); diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index dc6212a..af60590 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -1,4 +1,5 @@ import { recordAgentReceipt } from '../provisioning/agent-events.js'; +import { EventEmitter } from 'node:events'; import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'; import { isIP } from 'node:net'; import { db } from '../db/index.js'; @@ -8,6 +9,16 @@ import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, Hos export const AGENT_PROTOCOL_VERSION = 1; export const HEARTBEAT_MAX_AGE_MS = 90_000; const now = () => new Date().toISOString(); +const workEvents = new EventEmitter(); +workEvents.setMaxListeners(0); +export function waitForHostWork(hostId: string, signal: AbortSignal, timeoutMs = 20_000): Promise { + return new Promise(resolve => { + const done = () => { clearTimeout(timer); workEvents.off(hostId, done); signal.removeEventListener('abort', done); resolve(); }; + const timer = setTimeout(done, timeoutMs); + workEvents.on(hostId, done); signal.addEventListener('abort', done, { once: true }); + if (signal.aborted) done(); + }); +} const digest = (token: string) => createHash('sha256').update(token).digest('hex'); function database() { const handle = db(); ensureHostSchema(handle); return handle; } type HostRow = { id: string; name: string; hostname: string; mode: HostMode; port_min: number; port_max: number; bind_address: string; public_address: string; minimum_free_bytes: number; created_at: string; enrolled_at: string | null; last_heartbeat: string | null; inventory: string; credential_active: number }; @@ -113,6 +124,7 @@ export function deleteHost(id: string): { deleted: boolean; operationId?: string ON CONFLICT(host_id) DO UPDATE SET operation_id=excluded.operation_id,token_hash=excluded.token_hash,stage='queued',requested_at=excluded.requested_at,updated_at=excluded.updated_at,cleaned_at=NULL,error=NULL`) .run(id, operationId, digest(token), timestamp, timestamp); handle.prepare("UPDATE ec_hosts SET mode='draining' WHERE id=?").run(id); + queueMicrotask(() => workEvents.emit(id)); return { deleted: false, operationId }; }).immediate(); } @@ -287,7 +299,7 @@ export function requireReadyHost(hostId: string, options: { libraryId?: string; type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string }; function operationFromRow(row: OperationRow): HostOperation { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, payload: JSON.parse(row.payload), state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; } export function getHostOperation(id: string): HostOperation | null { const row = database().prepare('SELECT * FROM ec_host_operations WHERE id=?').get(id) as OperationRow | undefined; return row ? operationFromRow(row) : null; } -export function listHostOperations(hostId: string, limit = 100): HostOperation[] { return (database().prepare('SELECT * FROM ec_host_operations WHERE host_id=? ORDER BY created_at DESC,rowid DESC LIMIT ?').all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); } +export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] { return (database().prepare(`SELECT ${summary ? `id,host_id,kind,instance_id,generation,state,error,created_at,updated_at,'{}' AS payload,NULL AS result` : '*'} FROM ec_host_operations WHERE host_id=? ORDER BY created_at DESC,rowid DESC LIMIT ?`).all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); } /** Heartbeats can tell an executing create to stop at its next safe command boundary. */ export function pendingCreationCancellations(hostId: string): string[] { return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations @@ -300,7 +312,7 @@ export function queueHostOperation(hostId: string, kind: string, payload: Record if (!/^[a-z][a-z0-9_.-]{0,79}$/.test(kind)) throw new Error('Invalid operation kind.'); if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.'); if (options.generation !== undefined && (!Number.isSafeInteger(options.generation) || options.generation < 1)) throw new Error('Instance generation must be a positive integer.'); - return database().transaction(() => { + const operation = database().transaction(() => { if (options.idempotencyKey) { const existing = database().prepare('SELECT * FROM ec_host_operations WHERE host_id=? AND idempotency_key=?').get(hostId, options.idempotencyKey) as OperationRow | undefined; if (existing) { @@ -315,8 +327,10 @@ export function queueHostOperation(hostId: string, kind: string, payload: Record database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,?,?)').run(id, 'pending', timestamp); return getHostOperation(id)!; })(); + queueMicrotask(() => workEvents.emit(hostId)); + return operation; } -export function pendingHostOperations(hostId: string): HostOperation[] { +export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'write' = 'all'): HostOperation[] { const host = requireHost(hostId); if (host.mode === 'disabled' || !host.connected) return []; // Older agents hold activity.lock across the entire delivered batch, including @@ -328,8 +342,12 @@ export function pendingHostOperations(hostId: string): HostOperation[] { WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return []; // Accepted/running/unknown commands keep the same immutable ID. The agent // consults its durable journal before acknowledging or executing a replay. + const readOnly = `(kind IN ('host.inspect','instance.inspect','files.read','files.list','files.stat','logs.read') + OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`; + const laneFilter = lane === 'all' ? '' : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}`; return database().transaction(() => { const rows = database().prepare(`SELECT * FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown') + ${laneFilter} AND (? OR state != 'pending' OR kind NOT IN ('instance.create','create')) AND (kind!='instance.cancel-create' OR ?) AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id @@ -342,7 +360,7 @@ export function pendingHostOperations(hostId: string): HostOperation[] { })(); } export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation { - return database().transaction(() => { + const operation = database().transaction(() => { requireHost(hostId); const operation = getHostOperation(id); if (!operation || operation.hostId !== hostId) throw new Error('Operation was not found for this host.'); @@ -367,6 +385,8 @@ export function reportHostOperation(hostId: string, id: string, input: { state: } return getHostOperation(id)!; })(); + queueMicrotask(() => workEvents.emit(`operation:${id}`)); + return operation; } export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise { const deadline = Date.now() + timeoutMs; @@ -375,7 +395,7 @@ export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise if (!operation) throw new Error('Host operation was not found.'); if (['completed','failed','unknown'].includes(operation.state)) return operation; if (Date.now() >= deadline) throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`); - await new Promise(resolve => setTimeout(resolve, 200)); + await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now()))); } } diff --git a/src/lib/server/integrations/plutonium.ts b/src/lib/server/integrations/plutonium.ts index 66227f6..fbb1a26 100644 --- a/src/lib/server/integrations/plutonium.ts +++ b/src/lib/server/integrations/plutonium.ts @@ -22,13 +22,18 @@ export function freshRelease(): number | null { export function runningVersions(): PlutoniumVersion[] { const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[]; const target = freshRelease(); + const byServer = new Map(versions.map(version => [version.server, version])); + const epochs = new Map((db().prepare(`SELECT server, MAX(at) AS at FROM ( + SELECT server, submitted_at AS at FROM server_deployments WHERE server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild') + UNION ALL SELECT server, requested_at AS at FROM restart_rollout) GROUP BY server`).all() as { server: string; at: string | null }[]).map(row => [row.server, row.at])); + const recovering = new Set((db().prepare('SELECT server FROM restart_rollout WHERE healthy_at IS NULL').all() as { server: string }[]).map(row => row.server)); return listServers().map(server => { - const value = versions.find(v => v.server === server.name) + const value = byServer.get(server.name) ?? { server: server.name, revision: null, checked_at: null, error: null }; const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN; - const epoch = latestGameSubmission(server.name); + const epoch = epochs.get(server.name); const stopped = server.desired_state === 'stopped'; - const valid = !value.error && age >= 0 && age < HOUR && (!epoch || value.checked_at! > epoch) && !restartRecoveryPending(server.name); + const valid = !value.error && age >= 0 && age < HOUR && (!epoch || value.checked_at! > epoch) && !recovering.has(server.name); const revision = valid && !stopped ? value.revision : null; return { ...value, revision, stopped, update_status: stopped ? 'stopped' : revision === null || target === null ? 'unknown' diff --git a/src/lib/server/performance.ts b/src/lib/server/performance.ts new file mode 100644 index 0000000..85f64c9 --- /dev/null +++ b/src/lib/server/performance.ts @@ -0,0 +1,13 @@ +import { monitorEventLoopDelay } from 'node:perf_hooks'; + +let started = false; +export function startPerformanceMonitor() { + if (started || process.env.EC_PERF_LOG !== 'true') return; + started = true; + const delay = monitorEventLoopDelay({ resolution: 20 }); delay.enable(); + setInterval(() => { + console.info('[performance]', { eventLoopP95Ms: Math.round(delay.percentile(95) / 1e6), + eventLoopMaxMs: Math.round(delay.max / 1e6), rssMiB: Math.round(process.memoryUsage().rss / 1048576) }); + delay.reset(); + }, 60_000).unref(); +} diff --git a/src/lib/server/pipeline/bus.ts b/src/lib/server/pipeline/bus.ts index 5ad85e8..6ddf95c 100644 --- a/src/lib/server/pipeline/bus.ts +++ b/src/lib/server/pipeline/bus.ts @@ -46,6 +46,7 @@ export type BetaStage = export type BetaState = 'running' | 'ready-promote' | 'promoted' | 'rolled-back'; export type PipelineEvent = + | { type: 'fleet-invalidated' } | { type: 'server-telemetry'; entries: GameTelemetry[] } | { type: 'deployment-opened'; deploymentId: string; action: string; servers: string[] } | { type: 'server-changed'; deploymentId: string; server: string; files: number } @@ -58,7 +59,7 @@ export type PipelineEvent = error: string | null; } | { type: 'sweep-tick'; frozen: boolean; reason?: string } - | { type: 'server-status'; entries: ServerStatusEntry[] } + | { type: 'server-status'; entries: ServerStatusEntry[]; delta?: boolean } | { type: 'beta-opened'; betaId: number; scriptIds: number[]; servers: string[] } | { type: 'beta-server-stage'; @@ -78,9 +79,17 @@ class Bus extends EventEmitter { } const bus = new Bus(); +let invalidationPending = false; +export function invalidateFleet(): void { + if (invalidationPending) return; + invalidationPending = true; + // Flush after nested database transactions have committed. + queueMicrotask(() => { invalidationPending = false; bus.emit('event', { type: 'fleet-invalidated' }); }); +} export function emit(event: PipelineEvent): void { bus.emit('event', event); + if (event.type === 'beta-opened' || event.type === 'beta-state' || event.type === 'beta-server-stage') invalidateFleet(); } export function subscribe(listener: (event: PipelineEvent) => void): () => void { diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index f93797e..33a69b7 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -6,10 +6,10 @@ import { emit } from './bus.js'; import type { ServerStatusEntry } from './bus.js'; import { rconStatusResult } from '../integrations/rcon.js'; import type { RconFailure } from '../integrations/rcon.js'; -import { getServer, listServers, setServerField, type ServerRow } from '../fleet/servers.js'; +import { getServer, listServers, setServerFields, type ServerRow } from '../fleet/servers.js'; import { komodoInspectHealthy } from '../integrations/komodo.js'; import { restartSettling } from '../provisioning/index.js'; -import { serverObservation } from '../fleet/status.js'; +import { serverObservation, observationReader } from '../fleet/status.js'; import { telemetryEvidence, latestGameSubmission } from '../fleet/status-evidence.js'; import { isProvisioning } from '../fleet/provisioning.js'; import { STATUS_REFRESH_MS } from '../../server-freshness.js'; @@ -76,6 +76,15 @@ function encodeError(reason: RconFailure, detail: string): string { const pendingStatus = new Map>(); const statusQueue: Array<() => Promise> = []; let activeStatus = 0; +const changedStatus = new Map(); +function publishStatus(entry: ServerStatusEntry) { + const scheduled = changedStatus.size > 0; + changedStatus.set(entry.server, entry); + if (!scheduled) queueMicrotask(() => { + const entries = [...changedStatus.values()]; changedStatus.clear(); + emit({ type: 'server-status', delta: true, entries }); + }); +} function drainStatusQueue(): void { while (activeStatus < STATUS_CONCURRENCY && statusQueue.length) { @@ -95,9 +104,9 @@ function queueStatus(server: string): Promise { let capacity: number | null = null; try { capacity = await refreshServerStatus(server); - // Consumers expect a full snapshot. Re-read it now so neither game - // data nor container evidence waits for the slowest fleet member. - emit({ type: 'server-status', entries: listServers().map(statusEntry) }); + // Publish this server's latest evidence without waiting for the fleet. + const current = getServer(server); + if (current) publishStatus(statusEntry(current)); } catch (err) { console.error(`[worker] status error for ${server}:`, err); } finally { pendingStatus.delete(server); finish(capacity); } }); @@ -113,8 +122,9 @@ export async function fanoutStatus(): Promise { try { // Metrics remain complete fleet collections, using freshness at capture // time rather than the status saved at the beginning of a long scan. + const observe = observationReader(); recordMetrics(listServers().map(srv => { - const current = serverObservation(srv); + const current = observe(srv); return { server: srv.name, game: srv.game ?? 'unknown', map: selectedZombieMap(srv.game, current.last_map ?? srv.map, srv.map_slug)?.friendly ?? current.last_map ?? srv.map ?? 'Unknown map', players: current.health === 'online' ? current.last_players : null, @@ -127,6 +137,8 @@ async function refreshServerStatus(server: string): Promise { const srv = getServer(server); if (!srv) return null; let capacity: number | null = null; + let updates: Partial> = {}; + const flush = () => { setServerFields(srv.name, updates); updates = {}; }; if (!isProvisioning(srv.name)) { const epoch = latestGameSubmission(srv.name); const unchanged = () => epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name) @@ -136,53 +148,56 @@ async function refreshServerStatus(server: string): Promise { : await komodoInspectHealthy(srv.name).catch(err => ({ ok: false as const, state: 'unknown', reason: (err as Error).message })); // A lifecycle request may be submitted while an inspection is in flight. if (unchanged()) { - setServerField(srv.name, 'container_state', container.ok ? 'running' : container.state?.toLowerCase() ?? 'unknown'); - setServerField(srv.name, 'container_checked_at', new Date().toISOString()); + updates.container_state = container.ok ? 'running' : container.state?.toLowerCase() ?? 'unknown'; + updates.container_checked_at = new Date().toISOString(); if (!container.ok) { - setServerField(srv.name, 'last_error', null); - setServerField(srv.name, 'last_reply_snippet', null); + updates.last_error = null; + updates.last_reply_snippet = null; } else if (!telemetryEvidence(srv.name)) { + flush(); const result = await rconStatusResult(srv.name).catch(err => ({ ok: false as const, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined })); if (unchanged()) { const now = new Date().toISOString(); - setServerField(srv.name, 'last_checked_at', now); - setServerField(srv.name, 'last_reply_snippet', result.rawReply ?? null); + updates.last_checked_at = now; + updates.last_reply_snippet = result.rawReply ?? null; if (result.ok) { capacity = result.status.capacity ?? null; - setServerField(srv.name, 'last_players', result.status.players); - setServerField(srv.name, 'last_map', result.status.map); - setServerField(srv.name, 'last_gametype', result.status.gametype); - setServerField(srv.name, 'last_roster', JSON.stringify(result.status.roster)); - setServerField(srv.name, 'last_seen_at', now); - setServerField(srv.name, 'last_error', null); - } else setServerField(srv.name, 'last_error', encodeError(result.reason, result.detail)); + updates.last_players = result.status.players; + updates.last_map = result.status.map; + updates.last_gametype = result.status.gametype; + updates.last_roster = JSON.stringify(result.status.roster); + updates.last_seen_at = now; + updates.last_error = null; + } else updates.last_error = encodeError(result.reason, result.detail); } } } } else { - setServerField(srv.name, 'last_error', null); - setServerField(srv.name, 'last_reply_snippet', null); + updates.last_error = null; + updates.last_reply_snippet = null; } + flush(); const current = getServer(srv.name) ?? srv; const status = serverObservation(current); // Occupancy timers and match clearing follow the same source as the UI. if (status.health === 'online' && status.last_players !== null) { if (status.last_players > 0) { - setServerField(srv.name, 'active_since', current.active_since ?? status.last_seen_at); - setServerField(srv.name, 'inactive_since', null); + updates.active_since = current.active_since ?? status.last_seen_at; + updates.inactive_since = null; } else { - setServerField(srv.name, 'active_since', null); - setServerField(srv.name, 'inactive_since', current.inactive_since ?? status.last_seen_at); - setServerField(srv.name, 'game_started_at', null); - setServerField(srv.name, 'game_ended_at', null); + updates.active_since = null; + updates.inactive_since = current.inactive_since ?? status.last_seen_at; + updates.game_started_at = null; + updates.game_ended_at = null; } } + flush(); return capacity; } -function statusEntry(srv: ServerRow): ServerStatusEntry { - const final = serverObservation(srv); +export function statusEntry(srv: ServerRow, observe = serverObservation): ServerStatusEntry { + const final = observe(srv); return { server: srv.name, players: final.last_players, map: final.last_map ?? srv.map, gametype: final.last_gametype, roster: final.roster, health: final.health, lastSeenAt: final.last_seen_at, lastError: final.last_error, replySnippet: final.replySnippet, diff --git a/src/lib/server/provisioning/progress.ts b/src/lib/server/provisioning/progress.ts index c02d375..125219d 100644 --- a/src/lib/server/provisioning/progress.ts +++ b/src/lib/server/provisioning/progress.ts @@ -1,4 +1,5 @@ import { db } from '../db/index.js'; +import { invalidateFleet } from '../pipeline/bus.js'; import { provisioningPhases, type ProvisioningPhase, type ProvisioningProgress, type ProvisioningCheck, type ProvisioningCheckpoint, type ProvisioningBlocker, type RecoveryObservation } from '../../provisioning-detail.js'; @@ -116,5 +117,6 @@ export function recordProgress(id: string, update: ProgressUpdate): void { const insert = handle.prepare(`INSERT INTO server_deployment_events (deployment_id, created_at, message, kind, phase_key, details_json) VALUES (?, ?, ?, ?, ?, ?)`); for (const event of events) insert.run(id, now, event.message, event.kind, next.phase, event.details ? JSON.stringify(event.details) : null); + if (events.length) invalidateFleet(); })(); } diff --git a/src/lib/server/reports/database.ts b/src/lib/server/reports/database.ts new file mode 100644 index 0000000..b99e65a --- /dev/null +++ b/src/lib/server/reports/database.ts @@ -0,0 +1,5 @@ +import Database from 'better-sqlite3'; +import { workerData } from 'node:worker_threads'; + +const handle = new Database(workerData.dbPath, { readonly: true, fileMustExist: true }); +export function db() { return handle; } diff --git a/src/lib/server/reports/index.ts b/src/lib/server/reports/index.ts new file mode 100644 index 0000000..b1301a3 --- /dev/null +++ b/src/lib/server/reports/index.ts @@ -0,0 +1,64 @@ +import { Worker } from 'node:worker_threads'; +import { resolve } from 'node:path'; +import { config } from '../config.js'; +import { db } from '../db/index.js'; +import type { MetricsQuery, MetricsData } from '../../metrics.js'; +import type { LeaderboardOptions, LeaderboardResponse, PlayerProfile, LeaderboardEntry } from '../../leaderboard.js'; + +class ReportWorker { + private worker: Worker | null = null; + private sequence = 0; + private pending = new Map void; reject: (error: Error) => void; timer: NodeJS.Timeout }>(); + private flights = new Map>(); + async close() { + const worker = this.worker; this.worker = null; + for (const pending of this.pending.values()) { clearTimeout(pending.timer); pending.reject(new Error('Report service stopped')); } + this.pending.clear(); this.flights.clear(); + await worker?.terminate(); + } + run(kind: string, input: unknown): Promise { + const requestInput = input as { query?: MetricsQuery; now?: number }; + const key = JSON.stringify(kind === 'metrics' ? [kind, requestInput.query, Math.floor(requestInput.now! / 30_000)] : [kind, input]); + const flight = this.flights.get(key); + if (flight) return flight; + if (this.pending.size >= 64) return Promise.reject(new Error('Reports are busy. Please retry shortly.')); + db(); // Finish migrations before opening the independent reader. + if (!this.worker) { + const worker = this.worker = new Worker(resolve(process.env.NODE_ENV === 'production' ? 'build/reports/worker.mjs' : '.svelte-kit/reports/worker.mjs'), { workerData: { dbPath: resolve(config.dbPath) } }); + worker.on('message', ({ id, value, error, elapsedMs, cached }) => { + const pending = this.pending.get(id); if (!pending) return; + clearTimeout(pending.timer); this.pending.delete(id); + if (process.env.EC_PERF_LOG === 'true') console.info('[report]', { kind, elapsedMs, cached }); + if (error) pending.reject(new Error(error)); else pending.resolve(value); + if (!this.pending.size) worker.unref(); + }); + const fail = (error: Error) => { + if (this.worker !== worker) return; + this.worker = null; + for (const pending of this.pending.values()) { clearTimeout(pending.timer); pending.reject(error); } + this.pending.clear(); void worker.terminate(); + }; + worker.on('error', fail); + worker.on('exit', () => fail(new Error('Report worker stopped. Please retry.'))); + } + this.worker.ref(); + const id = ++this.sequence; + const promise = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + // Reset the worker rather than leaving an expensive abandoned query queued. + const worker = this.worker; this.worker = null; + for (const item of this.pending.values()) { clearTimeout(item.timer); item.reject(new Error('Report timed out. Please retry.')); } + this.pending.clear(); void worker?.terminate(); + }, 60_000); + this.pending.set(id, { resolve, reject, timer }); this.worker!.postMessage({ id, kind, input }); + }).finally(() => { if (this.flights.get(key) === promise) this.flights.delete(key); }); + this.flights.set(key, promise); + return promise; + } +} +const metrics = new ReportWorker(); +const leaderboard = new ReportWorker(); +export const metricsReport = (query: MetricsQuery) => metrics.run('metrics', { query, now: Date.now() }); +export const leaderboardReport = (query: LeaderboardOptions) => leaderboard.run('leaderboard', query); +export const playerReport = (guid: string, offset: number, options: LeaderboardOptions) => leaderboard.run<{ profile: PlayerProfile | null; entry: LeaderboardEntry | null }>('player', { guid, offset, options }); +export async function closeReports() { await Promise.all([metrics.close(), leaderboard.close()]); } diff --git a/src/lib/server/reports/revisions.ts b/src/lib/server/reports/revisions.ts new file mode 100644 index 0000000..64cbcc7 --- /dev/null +++ b/src/lib/server/reports/revisions.ts @@ -0,0 +1,14 @@ +import type Database from 'better-sqlite3'; + +export function installReportRevisions(handle: Database.Database) { + handle.exec(`CREATE TABLE IF NOT EXISTS report_revisions (topic TEXT PRIMARY KEY, revision INTEGER NOT NULL DEFAULT 0); + INSERT OR IGNORE INTO report_revisions(topic) VALUES ('metrics'), ('leaderboard');`); + for (const [topic, tables] of Object.entries({ metrics: ['metric_samples', 'servers'], + leaderboard: ['leaderboard_matches', 'leaderboard_match_players', 'leaderboard_players'] })) { + for (const table of tables) for (const action of ['INSERT', 'UPDATE', 'DELETE']) { + const condition = table === 'servers' && action === 'UPDATE' ? 'WHEN OLD.destroyed_at IS NOT NEW.destroyed_at OR OLD.name IS NOT NEW.name' : ''; + handle.exec(`CREATE TRIGGER IF NOT EXISTS report_${table}_${action} AFTER ${action} ON ${table} ${condition} + BEGIN UPDATE report_revisions SET revision=revision+1 WHERE topic='${topic}'; END;`); + } + } +} diff --git a/src/lib/server/reports/worker.ts b/src/lib/server/reports/worker.ts new file mode 100644 index 0000000..960ec21 --- /dev/null +++ b/src/lib/server/reports/worker.ts @@ -0,0 +1,33 @@ +import { parentPort } from 'node:worker_threads'; +import { db } from '../db/index.js'; +import { readMetrics } from '../fleet/metrics.js'; +import { getLeaderboard, getPlayerLeaderboardEntry } from '../fleet/leaderboard.js'; +import { getPlayerProfile } from '../fleet/player-profile.js'; + +const cache = new Map(); +let bytes = 0; +parentPort!.on('message', ({ id, kind, input }) => { + const started = performance.now(); + try { + const result = db().transaction(() => { + const topic = kind === 'metrics' ? 'metrics' : 'leaderboard'; + const { revision } = db().prepare('SELECT revision FROM report_revisions WHERE topic=?').get(topic) as { revision: number }; + const now = kind === 'metrics' ? input.now : Date.now(); + const key = JSON.stringify([kind, revision, kind === 'metrics' ? Math.floor(now / 30_000) : 0, kind === 'metrics' ? input.query : input]); + const saved = cache.get(key); + if (saved) { cache.delete(key); cache.set(key, saved); return { value: saved.value, cached: true }; } + const value = kind === 'metrics' ? readMetrics(input.query, now) + : kind === 'leaderboard' ? getLeaderboard(input) + : { profile: getPlayerProfile(input.guid, input.offset, input.options), entry: getPlayerLeaderboardEntry(input.guid, input.options) }; + const size = Buffer.byteLength(JSON.stringify(value)); + if (size < 16 * 1024 * 1024) { + cache.set(key, { value, bytes: size }); bytes += size; + while (cache.size > 64 || bytes > 32 * 1024 * 1024) { + const oldest = cache.keys().next().value!; bytes -= cache.get(oldest)!.bytes; cache.delete(oldest); + } + } + return { value, cached: false }; + })(); + parentPort!.postMessage({ id, ...result, elapsedMs: performance.now() - started }); + } catch (cause) { parentPort!.postMessage({ id, error: cause instanceof Error ? cause.message : 'Report failed' }); } +}); diff --git a/src/lib/ui/campaigns.ts b/src/lib/ui/campaigns.ts new file mode 100644 index 0000000..b773519 --- /dev/null +++ b/src/lib/ui/campaigns.ts @@ -0,0 +1,15 @@ +/** Load only the requested history, respecting the API's per-request page bound. */ +export async function fetchCampaigns(filters: Record, count: number, signal?: AbortSignal): Promise<{ deployments: T[]; total: number }> { + const deployments: T[] = []; + let total = Infinity; + while (deployments.length < Math.min(count, total)) { + const params = new URLSearchParams({ ...filters, limit: String(Math.min(500, count - deployments.length)), offset: String(deployments.length) }); + const response = await fetch(`/api/campaigns?${params}`, { signal }); + if (!response.ok) throw new Error('Could not refresh campaigns.'); + const page = await response.json() as { deployments: T[]; total: number }; + total = page.total; + deployments.push(...page.deployments); + if (!page.deployments.length) break; + } + return { deployments, total: Number.isFinite(total) ? total : 0 }; +} diff --git a/src/lib/ui/poll.ts b/src/lib/ui/poll.ts index a1409c9..50301dd 100644 --- a/src/lib/ui/poll.ts +++ b/src/lib/ui/poll.ts @@ -1,25 +1,32 @@ /** Poll visible page content without overlapping requests. Fleet watches use their own monitor. */ -export function pollVisible(task: () => void | Promise, interval: number, immediate = true): () => void { +export function pollVisible(task: (signal: AbortSignal) => unknown | Promise, interval: number, immediate = true): () => void { let stopped = false; let running = false; let timer: ReturnType | undefined; + let controller: AbortController | undefined; + let failures = 0; const visible = () => document.visibilityState !== 'hidden'; const run = async () => { if (stopped || running || !visible()) return; clearTimeout(timer); running = true; - try { await task(); } - catch (error) { console.error('Page refresh failed', error); } + controller = new AbortController(); + const timeout = setTimeout(() => controller?.abort(), 30_000); + try { failures = await task(controller.signal) === false ? failures + 1 : 0; } + catch (error) { if (!controller.signal.aborted) { failures++; console.error('Page refresh failed', error); } } finally { + clearTimeout(timeout); + controller = undefined; running = false; - if (!stopped && visible()) timer = setTimeout(run, interval); + if (!stopped && visible()) timer = setTimeout(run, interval * Math.min(8, 2 ** failures)); } }; const visibility = () => { clearTimeout(timer); + if (!visible()) controller?.abort(); if (document.visibilityState !== 'hidden') void run(); }; document.addEventListener('visibilitychange', visibility); if (immediate) void run(); else timer = setTimeout(run, interval); - return () => { stopped = true; clearTimeout(timer); document.removeEventListener('visibilitychange', visibility); }; + return () => { stopped = true; clearTimeout(timer); controller?.abort(); document.removeEventListener('visibilitychange', visibility); }; } diff --git a/src/lib/ui/stores/fleet.ts b/src/lib/ui/stores/fleet.ts index 5733645..f68832e 100644 --- a/src/lib/ui/stores/fleet.ts +++ b/src/lib/ui/stores/fleet.ts @@ -1,7 +1,9 @@ +import { freshStatusTimestamp } from '$lib/server-freshness.js'; +import { pollVisible } from '../poll.js'; import { selectServerObservation } from '$lib/server-observation.js'; -import { derived, writable } from 'svelte/store'; +import { derived, writable, get } from 'svelte/store'; import { browser } from '$app/environment'; -import { statusByServer, statusReceivedAt, syncServerTime, matchEventsByServer, telemetryByServer, uiNow } from './servers.js'; +import { statusByServer, statusReceivedAt, syncServerTime, serverTimeNow, matchEventsByServer, telemetryByServer, uiNow } from './servers.js'; import { type FleetServer, type FleetOperation } from '$lib/fleet-overview.js'; export const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({ @@ -11,9 +13,30 @@ export const fleetError = writable(''); export const fleetRefreshing = writable(false); export const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null); -export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, uiNow], ([snapshot, live, received, matches, reports, now]) => - snapshot.servers.map(server => { - const status = received > snapshot.receivedAt ? live.get(server.name) : undefined; +const observed = new Map(); +const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow], ([snapshot, live, reports, now]) => { + const times = snapshot.servers.flatMap(server => { + const status = live.get(server.name); + return [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at, status?.telemetry?.receivedAt, + server.last_seen_at, server.container_checked_at, server.last_checked_at, server.rcon_status?.last_seen_at, + server.telemetry?.receivedAt, reports.get(server.name)?.receivedAt]; + }); + return times.map(time => `${freshStatusTimestamp(time ?? null, now)}:${now - Date.parse(time ?? '') >= 0 && now - Date.parse(time ?? '') <= 120_000}`).join(','); +}); +export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => { + const now = get(uiNow); + const next = snapshot.servers.map(server => { + const candidate = live.get(server.name); + const status = (candidate?.receivedAt ?? received) > snapshot.receivedAt ? candidate : undefined; + const match = matches.get(server.name); + const report = reports.get(server.name); + const times = [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at, + status?.telemetry?.receivedAt, server.last_seen_at, server.container_checked_at, server.last_checked_at, + server.rcon_status?.last_seen_at, server.telemetry?.receivedAt, report?.receivedAt]; + const phase = times.map(time => `${freshStatusTimestamp(time ?? null, now)}:${now - Date.parse(time ?? '') >= 0 && now - Date.parse(time ?? '') <= 120_000}`).join(','); + const saved = observed.get(server.name); + if (saved && saved.server === server && saved.status === status && saved.match === match && saved.report === report && saved.phase === phase) return saved.value; + const remember = (value: FleetServer) => { observed.set(server.name, { server, status, match, report, phase, value }); return value; }; const base = status ? { ...server, last_players: status.players, last_map: status.map, last_gametype: status.gametype, last_seen_at: status.lastSeenAt, last_error: status.lastError, roster: status.roster, health: status.health, endpoint: status.endpoint, replySnippet: status.replySnippet, active_since: status.activeSince, @@ -23,38 +46,65 @@ export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, ma telemetry_after: status.telemetryAfter ?? server.telemetry_after, rcon_status: status.rconStatus, telemetry: status.telemetry ?? server.telemetry } : server; const current = selectServerObservation(base, reports.get(server.name), now); - const match = matches.get(server.name); // Apply lifecycle events immediately, until a newer status observes them // or confirms that the lobby has emptied. if (!match || current.health !== 'online' || (current.telemetry_after && Date.parse(match.receivedAt) <= Date.parse(current.telemetry_after)) || Date.parse(match.receivedAt) <= Date.parse(current.last_seen_at ?? '') || match.receivedAt < (current.game_started_at ?? '') - || match.receivedAt < (current.game_ended_at ?? '')) return current; - return { ...current, game_started_at: match.event === 'game_start' ? match.receivedAt : current.game_started_at, - game_ended_at: match.event === 'game_end' ? match.receivedAt : null }; - })); + || match.receivedAt < (current.game_ended_at ?? '')) return remember(current); + return remember({ ...current, game_started_at: match.event === 'game_start' ? match.receivedAt : current.game_started_at, + game_ended_at: match.event === 'game_end' ? match.receivedAt : null }); + }); + const names = new Set(snapshot.servers.map(server => server.name)); + for (const name of observed.keys()) if (!names.has(name)) observed.delete(name); + return next; +}); + + +let snapshotServerNow = 0; +export function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable>; serverNow: number }) { + if (!browser || snapshot.serverNow < snapshotServerNow) return; + snapshotServerNow = snapshot.serverNow; + syncServerTime(snapshot.serverNow); + fleetIntegrations.set(snapshot.integrations); + fleet.set({ servers: snapshot.servers, operations: snapshot.operations, receivedAt: Date.now() - Math.max(0, serverTimeNow() - snapshot.serverNow) }); +} +function getIntegrations() { return get(fleetIntegrations); } + + let inFlight: Promise | null = null; -export function refreshFleet(): Promise { +let refreshRequested = false; +export function refreshFleet(afterCurrent = false): Promise { if (!browser) return Promise.resolve(); - if (inFlight) return inFlight; + if (inFlight) { refreshRequested ||= afterCurrent; return inFlight; } inFlight = (async () => { fleetRefreshing.set(true); const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), 12_000); const requestedAt = Date.now(); try { - const responses = await Promise.all([ - fetch('/api/servers', { signal: controller.signal }), - fetch('/api/fleet', { signal: controller.signal }) + const results = await Promise.allSettled([ + (async () => { + const response = await fetch('/api/servers', { signal: controller.signal }); + if (!response.ok) throw new Error('Could not refresh servers. Retrying automatically.'); + const servers = await response.json(); + if (servers.serverNow < snapshotServerNow) return; + snapshotServerNow = servers.serverNow; + syncServerTime(servers.serverNow); fleetIntegrations.set(servers.integrations); + fleet.update(value => ({ ...value, servers: servers.servers, receivedAt: requestedAt })); + })(), + (async () => { + const response = await fetch('/api/fleet', { signal: controller.signal }); + if (!response.ok) throw new Error('Could not refresh operations. Retrying automatically.'); + const operations = await response.json(); + fleet.update(value => ({ ...value, operations: operations.operations })); + })() ]); - if (responses.some(response => !response.ok)) throw new Error('Could not refresh Fleet. Retrying automatically.'); - const [servers, operations] = await Promise.all(responses.map(response => response.json())); - syncServerTime(servers.serverNow); + const failure = results.find(result => result.status === 'rejected'); + if (failure?.status === 'rejected') throw failure.reason; fleetError.set(''); - fleetIntegrations.set(servers.integrations); - fleet.set({ servers: servers.servers, operations: operations.operations, receivedAt: requestedAt }); } catch (error) { fleetError.set(error instanceof Error && error.name !== 'AbortError' ? error.message : 'Fleet refresh timed out. Retrying automatically.'); @@ -62,23 +112,25 @@ export function refreshFleet(): Promise { clearTimeout(timeout); fleetRefreshing.set(false); } - })().finally(() => { inFlight = null; }); + })().finally(() => { + inFlight = null; + if (refreshRequested) { refreshRequested = false; void refreshFleet(); } + }); return inFlight; } let stop: (() => void) | null = null; export function startFleetMonitor(): () => void { if (!browser || stop) return () => {}; - void refreshFleet(); - const interval = setInterval(() => { - if (document.visibilityState !== 'hidden') void refreshFleet(); - }, 10_000); - const onFocus = () => { if (document.visibilityState === 'visible') void refreshFleet(); }; - document.addEventListener('visibilitychange', onFocus); + const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000); + let refreshTimer: ReturnType | undefined; + const changed = () => { + if (refreshTimer || document.visibilityState === 'hidden') return; + refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250); + }; + window.addEventListener('ec:fleet-changed', changed); stop = () => { - clearInterval(interval); - document.removeEventListener('visibilitychange', onFocus); - stop = null; + stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null; }; return stop; } diff --git a/src/lib/ui/stores/pipeline.ts b/src/lib/ui/stores/pipeline.ts index 0752505..f883d19 100644 --- a/src/lib/ui/stores/pipeline.ts +++ b/src/lib/ui/stores/pipeline.ts @@ -26,6 +26,9 @@ function updateDeployment(id: string, mutate: (d: PipelineDeployment) => void): export function startPipelineStream(): (() => void) | undefined { if (!browser || source) return; source = new EventSource('/api/events/pipeline'); + for (const event of ['server-changed', 'server-state', 'deployment-opened', 'fleet-invalidated']) { + source.addEventListener(event, () => window.dispatchEvent(new Event('ec:fleet-changed'))); + } source.addEventListener('deployment-opened', (evt) => { const data = JSON.parse((evt as MessageEvent).data) as { deploymentId: string; action: string; servers: string[] }; diff --git a/src/lib/ui/stores/servers.ts b/src/lib/ui/stores/servers.ts index c3e0bb5..b264133 100644 --- a/src/lib/ui/stores/servers.ts +++ b/src/lib/ui/stores/servers.ts @@ -17,6 +17,7 @@ export interface RconEndpoint { } export interface ServerStatus { + receivedAt?: number; server: string; containerState?: string | null; containerCheckedAt?: string | null; @@ -91,17 +92,17 @@ export function startServerStream(): (() => void) | undefined { }); }); source.addEventListener('server-status', (evt) => { - const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number }; + const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean }; syncServerTime(data.serverNow); - statusByServer.update(() => { - const m = new Map(); - for (const e of data.entries) m.set(e.server, e); + statusByServer.update(previous => { + const m = data.delta ? new Map(previous) : new Map(); + for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() }); return m; }); statusReceivedAt.set(Date.now()); }); if (tickInterval === null) { - tickInterval = setInterval(() => _uiNow.set(serverTimeNow()), 1000); + tickInterval = setInterval(() => { if (typeof document === 'undefined' || document.visibilityState !== 'hidden') _uiNow.set(serverTimeNow()); }, 1000); } return () => { source?.close(); source = null; if (tickInterval) clearInterval(tickInterval); tickInterval = null; }; } diff --git a/src/routes/+layout.svelte b/src/routes/+layout.svelte index 5863dd6..74809cb 100644 --- a/src/routes/+layout.svelte +++ b/src/routes/+layout.svelte @@ -1,10 +1,10 @@ diff --git a/src/routes/hosts/+page.server.ts b/src/routes/hosts/+page.server.ts new file mode 100644 index 0000000..2619534 --- /dev/null +++ b/src/routes/hosts/+page.server.ts @@ -0,0 +1,4 @@ +import { listHosts } from '$lib/server/hosts/index.js'; +import { presentHostPreparation } from '$lib/server/hosts/preparation.js'; + +export function load() { return { hosts: listHosts().map(presentHostPreparation) }; } diff --git a/src/routes/hosts/+page.svelte b/src/routes/hosts/+page.svelte index 2a536b5..a798e30 100644 --- a/src/routes/hosts/+page.svelte +++ b/src/routes/hosts/+page.svelte @@ -1,22 +1,24 @@ Hosts · Electric Cherry diff --git a/src/routes/hosts/[id]/+page.svelte b/src/routes/hosts/[id]/+page.svelte index d92218b..1ac2c83 100644 --- a/src/routes/hosts/[id]/+page.svelte +++ b/src/routes/hosts/[id]/+page.svelte @@ -1,5 +1,7 @@ {host?.name ?? 'Host'} · Electric Cherry diff --git a/src/routes/leaderboard/+page.server.ts b/src/routes/leaderboard/+page.server.ts new file mode 100644 index 0000000..863cd38 --- /dev/null +++ b/src/routes/leaderboard/+page.server.ts @@ -0,0 +1,10 @@ +import { GET } from '../api/leaderboard/+server.js'; +import { error } from '@sveltejs/kit'; +import type { LeaderboardResponse } from '$lib/leaderboard.js'; + +export async function load({ url }: { url: URL }) { + const response = await GET({ url }); + const body = await response.json(); + if (!response.ok) error(response.status, body.error); + return { leaderboard: body as LeaderboardResponse }; +} diff --git a/src/routes/leaderboard/+page.svelte b/src/routes/leaderboard/+page.svelte index 44ad8cb..3bb8c58 100644 --- a/src/routes/leaderboard/+page.svelte +++ b/src/routes/leaderboard/+page.svelte @@ -1,8 +1,10 @@