Compare commits

...
9 Commits
37 changed files with 1361 additions and 249 deletions
+31 -5
View File
@@ -32,7 +32,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
VERSION = "1.4.10" VERSION = "1.4.11"
LABEL = "io.electric-cherry" LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024 MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000 MANIFEST_MAX_ENTRIES = 10000
@@ -918,6 +918,8 @@ class Agent:
self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
self.library_progress = {} self.library_progress = {}
self.library_transfers = {} self.library_transfers = {}
self.inventory_manifest_cache = {}
self.inventory_manifest_lock = threading.Lock()
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
self.executing_id = None self.executing_id = None
@@ -1191,6 +1193,25 @@ class Agent:
continue continue
return samples return samples
def inventory_manifest(self, marker):
# Inventory only: launch/download verification must still read disk.
# Include file identity so atomic replacements invalidate even when
# their size and modification time match the previous manifest.
with self.inventory_manifest_lock:
try:
metadata = marker.stat()
signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
cached = self.inventory_manifest_cache.get(marker)
if cached is not None and cached[0] == signature:
return cached[1]
self.inventory_manifest_cache.pop(marker, None)
value = json.loads(marker.read_text())
self.inventory_manifest_cache[marker] = (signature, value)
return value
except (OSError, ValueError):
self.inventory_manifest_cache.pop(marker, None)
raise
def inventory(self): def inventory(self):
cpu_usage = self.cpu_usage_percent() cpu_usage = self.cpu_usage_percent()
usage = shutil.disk_usage(self.root) usage = shutil.disk_usage(self.root)
@@ -1210,13 +1231,18 @@ class Agent:
libraries, caches = [], [] libraries, caches = [], []
progress = {key: value.copy() for key, value in list(self.library_progress.items())} progress = {key: value.copy() for key, value in list(self.library_progress.items())}
progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())}) progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
for marker in (self.root / "libraries").glob("*/*/manifest.json"): manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
with self.inventory_manifest_lock:
for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
del self.inventory_manifest_cache[marker]
for marker in manifests:
# A crash between writing a manifest and its atomic promotion must # A crash between writing a manifest and its atomic promotion must
# never expose the staging directory as a ready library. # never expose the staging directory as a ready library.
if marker.parent.name.startswith("."): if marker.parent.name.startswith("."):
continue continue
try: try:
value = json.loads(marker.read_text()) value = self.inventory_manifest(marker)
# Rechecking an existing cache must revoke readiness until its # Rechecking an existing cache must revoke readiness until its
# entire contents have passed verification again. # entire contents have passed verification again.
if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists(): if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
@@ -1233,9 +1259,9 @@ class Agent:
except (OSError, ValueError, KeyError): except (OSError, ValueError, KeyError):
continue continue
ready = {item["id"] for item in libraries} ready = {item["id"] for item in libraries}
for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): for marker in partial_manifests:
try: try:
value = json.loads(marker.read_text()) value = self.inventory_manifest(marker)
library_id = identifier(value["libraryId"]) library_id = identifier(value["libraryId"])
if library_id in ready or library_id in progress: if library_id in ready or library_id in progress:
continue continue
+73
View File
@@ -386,6 +386,68 @@ class AgentTests(unittest.TestCase):
self.agent.journal.close() self.agent.journal.close()
self.temp.cleanup() self.temp.cleanup()
def test_inventory_reuses_manifests_until_mtime_changes(self):
marker = self.library / "manifest.json"
saved = json.loads(marker.read_text())
original_read = Path.read_text
reads = []
def read(path, *args, **kwargs):
if path == marker:
reads.append(path)
return original_read(path, *args, **kwargs)
with patch.object(Path, "read_text", read):
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
self.assertEqual(len(reads), 1)
metadata = marker.stat()
marker.write_text(json.dumps({**saved, "size": 8}))
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000))
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
self.assertEqual(len(reads), 2)
def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self):
marker = self.library / "manifest.json"
saved = json.loads(marker.read_text())
self.agent.inventory()
metadata = marker.stat()
module.atomic_json(marker, {**saved, "size": 8})
module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
self.assertEqual(marker.stat().st_size, metadata.st_size)
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
def test_inventory_discards_invalid_and_removed_manifests(self):
marker = self.library / "manifest.json"
saved = json.loads(marker.read_text())
self.agent.inventory()
marker.write_text("invalid JSON")
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
module.atomic_json(marker, saved)
self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1)
marker.unlink()
self.assertEqual(self.agent.inventory()["libraryCaches"], [])
self.assertNotIn(marker, self.agent.inventory_manifest_cache)
module.atomic_json(marker, {**saved, "size": 8})
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self):
self.agent.inventory()
progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3}
module.atomic_json(self.library / "transfer.json", progress)
self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing")
module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5})
cache = self.agent.inventory()["libraryCaches"][0]
self.assertEqual(cache["status"], "failed")
self.assertEqual(cache["bytesVerified"], 5)
(self.library / "content/game.ff").write_bytes(b"bad-data!")
with self.assertRaises(AgentError):
self.agent.verify_library(self.library)
(self.library / "manifest.json").write_text("invalid JSON")
with self.assertRaises(ValueError):
self.agent.verify_library(self.library)
def test_library_remove_preserves_other_versions_and_is_idempotent(self): def test_library_remove_preserves_other_versions_and_is_idempotent(self):
other = self.agent.library_dir("t5", "other-library") other = self.agent.library_dir("t5", "other-library")
other.mkdir() other.mkdir()
@@ -1179,6 +1241,17 @@ class AgentTests(unittest.TestCase):
self.assertEqual(caches[0]["status"], "missing") self.assertEqual(caches[0]["status"], "missing")
self.assertEqual(caches[0]["bytesDownloaded"], len(content)) self.assertEqual(caches[0]["bytesDownloaded"], len(content))
self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"]) self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"])
(staging / "content/zone/test.ff").write_bytes(content[:5])
original_read = Path.read_text
def read(path, *args, **kwargs):
if path == staging / "source.json":
self.fail("Unchanged partial manifest should be cached")
return original_read(path, *args, **kwargs)
with patch.object(Path, "read_text", read):
partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
self.assertEqual(partial["bytesDownloaded"], 5)
self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3} self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3}
caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"] caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
self.assertEqual(len(caches), 1) self.assertEqual(len(caches), 1)
+4
View File
@@ -595,6 +595,10 @@ missing, stale, or predates lifecycle changes. RCON status queries share one
operation per host, with two concurrent local probes and a 20-second batch operation per host, with two concurrent local probes and a 20-second batch
deadline. Individual failures remain visible per server. Both additions are deadline. Individual failures remain visible per server. Both additions are
capability-gated; older agents retain individual inspection and RCON operations. capability-gated; older agents retain individual inspection and RCON operations.
Agent 1.4.11 caches parsed library manifests during inventory collection until
their modification time, size, or file identity changes. Deleted manifests leave
the cache, while transfer progress and partial-file sizes are sampled each time.
Launch and download integrity verification continue to read from disk.
Agent 1.4.5 reserves two observation workers for instance health and read-only RCON, Agent 1.4.5 reserves two observation workers for instance health and read-only RCON,
and one for files, logs, and host inventory. Background reads arrive in batches of and one for files, logs, and host inventory. Background reads arrive in batches of
up to four. File receipt uploads release the instance lock before contacting the up to four. File receipt uploads release the instance lock before contacting the
+5 -4
View File
@@ -3,9 +3,10 @@ import { migrate } from '$lib/server/db/migrate.js';
import { startWorker } from '$lib/server/pipeline/worker.js'; import { startWorker } from '$lib/server/pipeline/worker.js';
import { interruptDeployments } from '$lib/server/provisioning/history.js'; import { interruptDeployments } from '$lib/server/provisioning/history.js';
import { authenticationHandle } from '$lib/server/auth/guard.js'; import { authenticationHandle } from '$lib/server/auth/guard.js';
import { emit } from '$lib/server/pipeline/bus.js'; import { invalidateFleet } from '$lib/server/pipeline/bus.js';
import { startPerformanceMonitor } from '$lib/server/performance.js'; import { startPerformanceMonitor } from '$lib/server/performance.js';
import { withOperationSource } from '$lib/server/hosts/operation-context.js'; import { withOperationSource } from '$lib/server/hosts/operation-context.js';
import { withHostRequestContext } from '$lib/server/hosts/request-context.js';
let booted = false; let booted = false;
function boot(): void { function boot(): void {
@@ -23,13 +24,13 @@ function boot(): void {
export const handle: Handle = async (input) => { export const handle: Handle = async (input) => {
boot(); boot();
const started = performance.now(); const started = performance.now();
const response = await withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`, const response = await withHostRequestContext(() => withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
() => authenticationHandle(input)); () => authenticationHandle(input)));
const elapsed = performance.now() - started; const elapsed = performance.now() - started;
response.headers.append('Server-Timing', `app;dur=${elapsed.toFixed(1)}`); 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/')) { 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) }); 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' }); if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) invalidateFleet();
return response; return response;
}; };
+15
View File
@@ -391,6 +391,21 @@ export function db(): Database.Database {
throw err; throw err;
} }
// Cache only runtime queries, after migrations have finished. Statements
// belong to this connection, so reopening the DB starts with an empty cache.
// Callers share statement options (e.g. raw()); use consistent options for
// identical SQL and pass values to run/get/all rather than binding permanently.
const statements = new Map<string, Database.Statement>();
const prepare = handle.prepare.bind(handle);
handle.prepare = ((sql: string) => {
let statement = statements.get(sql);
if (!statement) {
statement = prepare(sql);
statements.set(sql, statement);
}
return statement;
}) as typeof handle.prepare;
_db = handle; _db = handle;
return _db; return _db;
} }
+30 -18
View File
@@ -2,19 +2,31 @@ import { actionLabels, type ProvisioningAction } from '../../provisioning.js';
import { db } from '../db/index.js'; import { db } from '../db/index.js';
import type { FleetOperation } from '../../fleet-overview.js'; import type { FleetOperation } from '../../fleet-overview.js';
/** Read persisted work, including recent completions for one-shot lobby watches. */ /** Fleet clients need only active work and unresolved failures. */
export function fleetOperations(now = Date.now()): FleetOperation[] { export function fleetOperations(now = Date.now()): FleetOperation[] {
return readOperations(now);
}
/** Recent completions are loaded only by the server's own page. */
export function serverOperations(server: string, now = Date.now()): FleetOperation[] {
return readOperations(now, server);
}
function readOperations(now: number, server?: string): FleetOperation[] {
const since = new Date(now - 86_400_000).toISOString(); const since = new Date(now - 86_400_000).toISOString();
const bindings = { server: server ?? null, since };
const scope = (column: string) => server === undefined ? '' : `AND ${column} = @server`;
const handle = db(); const handle = db();
const operations: FleetOperation[] = []; const operations: FleetOperation[] = [];
const slots = handle.prepare('SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL') const slots = handle.prepare(`SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL ${scope('server')}`)
.all() as { server: string; requested_at: string; error: string | null }[]; .all(bindings) as { server: string; requested_at: string; error: string | null }[];
const recovering = new Map(slots.map(slot => [slot.server, slot])); const recovering = new Map(slots.map(slot => [slot.server, slot]));
const representedRecovery = new Set<string>(); const representedRecovery = new Set<string>();
const campaigns = handle.prepare(`SELECT d.id, d.action, d.started_at, ds.server, const campaigns = handle.prepare(`SELECT d.id, d.action, d.started_at, ds.server,
ds.restart_state, ds.error, ds.restarted_at, ds.players ds.restart_state, ds.error, ds.restarted_at, ds.players
FROM deployments d JOIN deployment_servers ds ON ds.deployment_id = d.id FROM deployments d JOIN deployment_servers ds ON ds.deployment_id = d.id
WHERE ds.server_deleted_at IS NULL AND (ds.restart_state != 'done' OR ds.checked_at >= ? OR d.finished_at >= ? OR d.started_at >= ?)`).all(since, since, since) as { WHERE ds.server_deleted_at IS NULL ${scope('ds.server')} AND (ds.restart_state != 'done'
${server === undefined ? '' : 'OR ds.checked_at >= @since OR d.finished_at >= @since OR d.started_at >= @since'})`).all(bindings) as {
id: string; action: string; started_at: string; server: string; restart_state: string; id: string; action: string; started_at: string; server: string; restart_state: string;
error: string | null; restarted_at: string | null; players: number | null; error: string | null; restarted_at: string | null; players: number | null;
}[]; }[];
@@ -34,19 +46,19 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
href: `/campaigns/${encodeURIComponent(row.id)}` }); href: `/campaigns/${encodeURIComponent(row.id)}` });
} }
// Only the newest lifecycle operation can still require action on a server. // Only the newest lifecycle operation can still require action on a server.
// Separate branches let SQLite use state and finish indexes; UNION deduplicates // Separate branches use state/rowid indexes for fleet reads and the server
// recent warnings that are also the newest operation. Unary + keeps terminal // index for detail history. Completed work is never loaded for the full fleet.
// branches on finished_at and rowid instead of searching historical states.
const lifecycle = handle.prepare(`SELECT id, server, action, state, stage, error, started_at FROM server_deployments const lifecycle = handle.prepare(`SELECT id, server, action, state, stage, error, started_at FROM server_deployments
WHERE server_deleted_at IS NULL AND state IN ('queued', 'running', 'waiting', 'awaiting-health') WHERE server_deleted_at IS NULL ${scope('server')} AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
${server === undefined ? '' : `UNION
SELECT id, server, action, state, stage, error, started_at FROM server_deployments
WHERE server_deleted_at IS NULL AND server = @server
AND +state IN ('done', 'warning', 'cancelled', 'failed', 'interrupted') AND finished_at >= @since`}
UNION UNION
SELECT id, server, action, state, stage, error, started_at FROM server_deployments SELECT id, server, action, state, stage, error, started_at FROM server_deployments
WHERE server_deleted_at IS NULL AND +state IN ('done', 'warning', 'cancelled') AND finished_at >= ? WHERE server_deleted_at IS NULL ${scope('server')} AND +state IN ('failed', 'interrupted')
UNION AND rowid IN (SELECT MAX(rowid) FROM server_deployments WHERE 1=1 ${scope('server')} GROUP BY server)`)
SELECT id, server, action, state, stage, error, started_at FROM server_deployments .all(bindings) as { id: string; server: string; action: string; state: string; stage: string;
WHERE server_deleted_at IS NULL AND +state IN ('failed', 'interrupted', 'warning')
AND rowid IN (SELECT MAX(rowid) FROM server_deployments GROUP BY server)`)
.all(since) as { id: string; server: string; action: string; state: string; stage: string;
error: string | null; started_at: string }[]; error: string | null; started_at: string }[];
for (const row of lifecycle) { for (const row of lifecycle) {
if (row.state === 'awaiting-health') representedRecovery.add(row.server); if (row.state === 'awaiting-health') representedRecovery.add(row.server);
@@ -59,8 +71,8 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
detail: row.error ?? row.stage, href: `/provisioning/${encodeURIComponent(row.id)}` }); detail: row.error ?? row.stage, href: `/provisioning/${encodeURIComponent(row.id)}` });
} }
const betas = handle.prepare(`SELECT b.id, b.opened_at, bs.server, bs.stage, bs.error const betas = handle.prepare(`SELECT b.id, b.opened_at, bs.server, bs.stage, bs.error
FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL AND b.state IN ('running', 'ready-promote')`) FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL ${scope('bs.server')} AND b.state IN ('running', 'ready-promote')`)
.all() as { id: number; opened_at: string; server: string; stage: string; error: string | null }[]; .all(bindings) as { id: number; opened_at: string; server: string; stage: string; error: string | null }[];
const stages: Record<string, string> = { queued: 'Waiting for an earlier test wave', deploying: 'Deploying beta', const stages: Record<string, string> = { queued: 'Waiting for an earlier test wave', deploying: 'Deploying beta',
'awaiting-empty': 'Waiting for an empty lobby', restarting: 'Restarting', 'awaiting-health': 'Recovering', 'awaiting-empty': 'Waiting for an empty lobby', restarting: 'Restarting', 'awaiting-health': 'Recovering',
'awaiting-match': 'Observing a match', ready: 'Ready for review', failed: 'Beta needs attention' }; 'awaiting-match': 'Observing a match', ready: 'Ready for review', failed: 'Beta needs attention' };
@@ -71,9 +83,9 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
state: row.stage === 'failed' ? 'attention' : 'active', label: stages[row.stage] ?? row.stage, state: row.stage === 'failed' ? 'attention' : 'active', label: stages[row.stage] ?? row.stage,
detail: row.error ?? recovering.get(row.server)?.error ?? `Beta ${row.id}`, href: `/betas/${row.id}` }); detail: row.error ?? recovering.get(row.server)?.error ?? `Beta ${row.id}`, href: `/betas/${row.id}` });
} }
const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL AND m.state = 'failed' const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL ${scope('m.server')} AND m.state = 'failed'
AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id)`) AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id)`)
.all() as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[]; .all(bindings) as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
for (const job of maintenance) { for (const job of maintenance) {
operations.push({ key: `maintenance:${job.id}`, group: `maintenance:${job.id}`, kind: 'maintenance', operations.push({ key: `maintenance:${job.id}`, group: `maintenance:${job.id}`, kind: 'maintenance',
server: job.server, state: 'attention', startedAt: job.requested_at, server: job.server, state: 'attention', startedAt: job.requested_at,
+18 -9
View File
@@ -1,26 +1,32 @@
import { db } from '../db/index.js'; import { db } from '../db/index.js';
import { listServers } from './servers.js'; import { getServer, listServers } from './servers.js';
import { observationReader } from './status.js'; import { observationReader, serverObservation } from './status.js';
import { config, komodoConfigured, rconConfigured } from '../config.js'; import { config, komodoConfigured, rconConfigured } from '../config.js';
import { runningVersions } from '../integrations/plutonium.js'; import { runningVersions } from '../integrations/plutonium.js';
import { listHosts } from '../hosts/index.js'; import { listHosts } from '../hosts/index.js';
import type { NativeInstance } from '../backends/native.js'; import type { NativeInstance } from '../backends/native.js';
import { serverOperations } from './overview.js';
export function fleetSnapshot() { export function fleetSnapshot(serverName?: string) {
const bindings = serverName === undefined ? [] : [serverName];
const rconHost = config.rcon.host; const rconHost = config.rcon.host;
const hosts = listHosts(); const hosts = listHosts();
const hostsById = new Map(hosts.map(({ inventory: _inventory, ...host }) => [host.id, host])); const hostsById = new Map(hosts.map(({ inventory: _inventory, ...host }) => [host.id, host]));
const versions = new Map(runningVersions().map(version => [version.server, version])); const versions = new Map(runningVersions(serverName).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 instances = new Map((db().prepare(`SELECT * FROM native_instances WHERE retired_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
.all(...bindings) as NativeInstance[]).map(instance => [instance.server, instance]));
const holds = new Map<string, { target: string; betaId: number }[]>(); const holds = new Map<string, { target: string; betaId: number }[]>();
for (const row of db().prepare(`SELECT bsv.server, s.target_rel_path AS target, b.id AS betaId 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 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 JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')
${serverName === undefined ? '' : 'AND bsv.server = ?'}`).all(...bindings) as
{ server: string; target: string; betaId: number }[]) { { 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 entries = holds.get(row.server) ?? []; entries.push({ target: row.target, betaId: row.betaId }); holds.set(row.server, entries);
} }
const observe = observationReader(); const observe = serverName === undefined ? observationReader() : serverObservation;
const servers = listServers().map((s) => { const selected = serverName === undefined ? undefined : getServer(serverName);
const rows = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
const servers = rows.map((s) => {
const instance = s.native_instance_id ? instances.get(s.name) : null; const instance = s.native_instance_id ? instances.get(s.name) : null;
return { return {
...observe(s), ...observe(s),
@@ -36,7 +42,7 @@ export function fleetSnapshot() {
return { return {
servers, servers,
serverNow: Date.now(), 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), reservedNames: serverName === undefined ? (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: { integrations: {
native: hosts.length > 0 || !komodoConfigured(), native: hosts.length > 0 || !komodoConfigured(),
hostsReady: hosts.filter(host => host.readiness === 'ready').length, hostsReady: hosts.filter(host => host.readiness === 'ready').length,
@@ -49,3 +55,6 @@ export function fleetSnapshot() {
}; };
} }
export function serverSnapshot(name: string) {
return { ...fleetSnapshot(name), operations: serverOperations(name) };
}
+12 -1
View File
@@ -332,9 +332,17 @@ function transferProgress(value: unknown, status: LibraryCacheState): LibraryTra
bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null }; bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null };
} }
type LibraryCacheMetadata = Pick<GameLibraryVersion, 'state' | 'sizeBytes' | 'manifestSha256'>;
const publishedCacheMetadata = new WeakMap<ReturnType<typeof database>, Map<string, LibraryCacheMetadata>>();
export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void { export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void {
if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.'); if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.');
const handle = database(); const handle = database();
// Published metadata is immutable. Keep caches local to the connection and
// avoid retaining a publication from an outer transaction that may roll back.
const versions = handle.inTransaction ? new Map<string, LibraryCacheMetadata>()
: publishedCacheMetadata.get(handle) ?? new Map<string, LibraryCacheMetadata>();
if (!handle.inTransaction) publishedCacheMetadata.set(handle, versions);
const now = new Date().toISOString(); const now = new Date().toISOString();
handle.transaction(() => { handle.transaction(() => {
const reported = new Set<string>(); const reported = new Set<string>();
@@ -344,7 +352,10 @@ export function updateHostLibraryCaches(hostId: string, input: unknown, options:
if (typeof id !== 'string') throw new Error('Invalid library version ID.'); if (typeof id !== 'string') throw new Error('Invalid library version ID.');
if (reported.has(id)) throw new Error('Duplicate version in host cache report.'); if (reported.has(id)) throw new Error('Duplicate version in host cache report.');
reported.add(id); reported.add(id);
const version = getLibraryVersion(id); if (!version || version.state !== 'ready') continue; const version = versions.get(id) ?? handle.prepare(`SELECT state, size_bytes AS sizeBytes, manifest_sha256 AS manifestSha256
FROM game_file_versions WHERE id=?`).get(id) as LibraryCacheMetadata | undefined;
if (!version || version.state !== 'ready') continue;
versions.set(id, version);
let status = raw.status ?? raw.state; let status = raw.status ?? raw.state;
if (!['missing', 'downloading', 'verifying', 'ready', 'failed', 'outdated'].includes(status)) throw new Error('Invalid library cache state.'); if (!['missing', 'downloading', 'verifying', 'ready', 'failed', 'outdated'].includes(status)) throw new Error('Invalid library cache state.');
const checksum = typeof raw.manifestSha256 === 'string' ? raw.manifestSha256 : null; const checksum = typeof raw.manifestSha256 === 'string' ? raw.manifestSha256 : null;
+10 -3
View File
@@ -6,6 +6,7 @@ import { db } from '../db/index.js';
import { ensureHostSchema } from './schema.js'; import { ensureHostSchema } from './schema.js';
import { successfulObservation, routineObservationSql, importantOperationSql, operationTargetSql, safeOperationTarget } from './observations.js'; import { successfulObservation, routineObservationSql, importantOperationSql, operationTargetSql, safeOperationTarget } from './observations.js';
import { currentOperationSource } from './operation-context.js'; import { currentOperationSource } from './operation-context.js';
import { parseHostInventory } from './request-context.js';
import { fleetSettings } from '../settings.js'; import { fleetSettings } from '../settings.js';
import { recordInstanceResources, validateInstanceResources } from './calibration.js'; import { recordInstanceResources, validateInstanceResources } from './calibration.js';
import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js'; import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
@@ -38,7 +39,7 @@ function presentDecommission(row: DecommissionRow | undefined): HostDecommission
} }
function hostFromRow(row: HostRow): Host { function hostFromRow(row: HostRow): Host {
const enabling = database().prepare('SELECT requested_at FROM ec_host_enable_requests WHERE host_id=?').get(row.id) as { requested_at: string } | undefined; const enabling = database().prepare('SELECT requested_at FROM ec_host_enable_requests WHERE host_id=?').get(row.id) as { requested_at: string } | undefined;
const inventory: HostInventory = JSON.parse(row.inventory); const inventory = parseHostInventory(row.id, row.inventory);
const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS; const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS;
const reasons: string[] = []; const reasons: string[] = [];
if (!row.enrolled_at) reasons.push('Agent has not enrolled.'); if (!row.enrolled_at) reasons.push('Agent has not enrolled.');
@@ -304,9 +305,15 @@ export function enrollHost(token: unknown, rawInventory: unknown): { hostId: str
} }
export function authenticateCredential(token: string): Host | null { export function authenticateCredential(token: string): Host | null {
if (!token || token.length > 200) return null; if (!token || token.length > 200) return null;
const row = database().prepare('SELECT id,host_id FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL').get(digest(token)) as { id: string; host_id: string } | undefined; const row = database().prepare('SELECT id,host_id,last_used_at FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL')
.get(digest(token)) as { id: string; host_id: string; last_used_at: string | null } | undefined;
if (!row) return null; if (!row) return null;
database().prepare('UPDATE ec_host_credentials SET last_used_at=? WHERE id=?').run(now(), row.id); const timestamp = Date.now();
if (row.last_used_at === null || timestamp - Date.parse(row.last_used_at) >= 60_000) {
database().prepare(`UPDATE ec_host_credentials SET last_used_at=? WHERE id=? AND revoked_at IS NULL
AND (last_used_at IS NULL OR last_used_at<=?)`)
.run(new Date(timestamp).toISOString(), row.id, new Date(timestamp - 60_000).toISOString());
}
return getHost(row.host_id); return getHost(row.host_id);
} }
export function revokeCredentials(hostId: string): void { export function revokeCredentials(hostId: string): void {
+28
View File
@@ -0,0 +1,28 @@
import { AsyncLocalStorage } from 'node:async_hooks';
import type { HostInventory } from '../../hosts.js';
type InventoryCache = Map<string, { json: string; inventory: HostInventory }>;
const requests = new AsyncLocalStorage<{ inventories: InventoryCache | null }>();
export function withHostRequestContext<T>(work: () => T | Promise<T>): Promise<T> {
const context = { inventories: new Map() as InventoryCache | null };
return requests.run(context, async () => {
try { return await work(); }
finally {
// Detached work can inherit this context. Release request data and
// stop caching when the response is ready, even if that work continues.
context.inventories = null;
}
});
}
export function parseHostInventory(hostId: string, json: string): HostInventory {
const cache = requests.getStore()?.inventories;
const saved = cache?.get(hostId);
if (saved?.json === json) return saved.inventory;
const inventory: HostInventory = JSON.parse(json);
// Compare the stored JSON so heartbeats and transaction rollbacks within
// a request remain visible. Other host state is always read afresh.
cache?.set(hostId, { json, inventory });
return inventory;
}
+10 -5
View File
@@ -20,13 +20,18 @@ export function freshRelease(): number | null {
const age = latest.checkedAt ? Date.now() - Date.parse(latest.checkedAt) : NaN; const age = latest.checkedAt ? Date.now() - Date.parse(latest.checkedAt) : NaN;
return !latest.error && age >= 0 && age < HOUR ? latest.revision : null; return !latest.error && age >= 0 && age < HOUR ? latest.revision : null;
} }
export function runningVersions(): PlutoniumVersion[] { export function runningVersions(serverName?: string): PlutoniumVersion[] {
const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[]; const bindings = serverName === undefined ? [] : [serverName];
const versions = db().prepare(`SELECT * FROM plutonium_versions ${serverName === undefined ? '' : 'WHERE server = ?'}`)
.all(...bindings) as PlutoniumVersion[];
const target = freshRelease(); const target = freshRelease();
const byServer = new Map(versions.map(version => [version.server, version])); const byServer = new Map(versions.map(version => [version.server, version]));
const epochs = latestGameSubmissions(); const epochs = serverName === undefined ? latestGameSubmissions() : new Map([[serverName, latestGameSubmission(serverName)]]);
const recovering = new Set((db().prepare('SELECT server FROM restart_rollout WHERE healthy_at IS NULL').all() as { server: string }[]).map(row => row.server)); const recovering = new Set((db().prepare(`SELECT server FROM restart_rollout WHERE healthy_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
return listServers().map(server => { .all(...bindings) as { server: string }[]).map(row => row.server));
const selected = serverName === undefined ? undefined : getServer(serverName);
const servers = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
return servers.map(server => {
const value = byServer.get(server.name) const value = byServer.get(server.name)
?? { server: server.name, revision: null, checked_at: null, error: null }; ?? { server: server.name, revision: null, checked_at: null, error: null };
const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN; const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN;
+4 -1
View File
@@ -7,6 +7,7 @@ import { pruneHostObservations } from '../hosts/index.js';
interface PendingRow { interface PendingRow {
deployment_id: string; server: string; restart_state: string; deployment_id: string; server: string; restart_state: string;
restarted_at: string | null; files_changed: number; restarted_at: string | null; files_changed: number;
players: number | null; error: string | null;
} }
type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number }; type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number };
let sweepInFlight: Promise<SweepResult> | null = null; let sweepInFlight: Promise<SweepResult> | null = null;
@@ -39,7 +40,9 @@ async function runSweep(): Promise<SweepResult> {
attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END, attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END,
restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`) restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`)
.run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server); .run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server);
emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error }); if (state !== row.restart_state || operation.players !== row.players || error !== row.error) {
emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error });
}
} }
db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS ( db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS (
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso()); SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso());
+156 -123
View File
@@ -6,131 +6,164 @@ import { browser } from '$app/environment';
import { statusByServer, statusReceivedAt, syncServerTime, serverTimeNow, 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'; import { type FleetServer, type FleetOperation } from '$lib/fleet-overview.js';
export const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({ // Each server detail view owns a scoped snapshot; it must not replace the shared fleet.
servers: [], operations: [], receivedAt: 0 export function createFleetStore(serverName?: string) {
}); const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({
export const fleetError = writable(''); servers: [], operations: [], receivedAt: 0
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);
const observed = new Map<string, { server: FleetServer; status: unknown; match: unknown; report: unknown; phase: string; value: FleetServer }>();
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(','); const fleetError = writable('');
}); const fleetRefreshing = writable(false);
export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => { const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null);
const now = get(uiNow);
const next = snapshot.servers.map(server => { const observed = new Map<string, { server: FleetServer; status: unknown; match: unknown; report: unknown; phase: string; value: FleetServer }>();
const candidate = live.get(server.name); const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow], ([snapshot, live, reports, now]) => {
const status = (candidate?.receivedAt ?? received) > snapshot.receivedAt ? candidate : undefined; const times = snapshot.servers.flatMap(server => {
const match = matches.get(server.name); const status = live.get(server.name);
const report = reports.get(server.name); return [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at, status?.telemetry?.receivedAt,
const times = [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at, server.last_seen_at, server.container_checked_at, server.last_checked_at, server.rcon_status?.last_seen_at,
status?.telemetry?.receivedAt, server.last_seen_at, server.container_checked_at, server.last_checked_at, server.telemetry?.receivedAt, reports.get(server.name)?.receivedAt];
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(','); return 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,
inactive_since: status.inactiveSince, game_started_at: status.gameStartedAt,
game_ended_at: status.gameEndedAt, container_state: status.containerState ?? server.container_state,
container_checked_at: status.containerCheckedAt ?? server.container_checked_at,
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);
// 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 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)); const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => {
for (const name of observed.keys()) if (!names.has(name)) observed.delete(name); const now = get(uiNow);
return next; 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);
let snapshotServerNow = 0; const report = reports.get(server.name);
export function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; serverNow: number }) { const times = [status?.lastSeenAt, status?.containerCheckedAt, status?.rconStatus?.last_seen_at,
if (!browser || snapshot.serverNow < snapshotServerNow) return; status?.telemetry?.receivedAt, server.last_seen_at, server.container_checked_at, server.last_checked_at,
snapshotServerNow = snapshot.serverNow; server.rcon_status?.last_seen_at, server.telemetry?.receivedAt, report?.receivedAt];
syncServerTime(snapshot.serverNow); const phase = times.map(time => `${freshStatusTimestamp(time ?? null, now)}:${now - Date.parse(time ?? '') >= 0 && now - Date.parse(time ?? '') <= 120_000}`).join(',');
fleetIntegrations.set(snapshot.integrations); const saved = observed.get(server.name);
fleet.set({ servers: snapshot.servers, operations: snapshot.operations, receivedAt: Date.now() - Math.max(0, serverTimeNow() - snapshot.serverNow) }); 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; };
function getIntegrations() { return get(fleetIntegrations); } 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,
inactive_since: status.inactiveSince, game_started_at: status.gameStartedAt,
let inFlight: Promise<void> | null = null; game_ended_at: status.gameEndedAt, container_state: status.containerState ?? server.container_state,
let refreshRequested = false; container_checked_at: status.containerCheckedAt ?? server.container_checked_at,
export function refreshFleet(afterCurrent = false): Promise<void> { telemetry_after: status.telemetryAfter ?? server.telemetry_after, rcon_status: status.rconStatus,
if (!browser) return Promise.resolve(); telemetry: status.telemetry ?? server.telemetry } : server;
if (inFlight) { refreshRequested ||= afterCurrent; return inFlight; } const current = selectServerObservation(base, reports.get(server.name), now);
inFlight = (async () => { // Apply lifecycle events immediately, until a newer status observes them
fleetRefreshing.set(true); // or confirms that the lobby has emptied.
const controller = new AbortController(); if (!match || current.health !== 'online'
const timeout = setTimeout(() => controller.abort(), 12_000); || (current.telemetry_after && Date.parse(match.receivedAt) <= Date.parse(current.telemetry_after))
const requestedAt = Date.now(); || Date.parse(match.receivedAt) <= Date.parse(current.last_seen_at ?? '')
try { || match.receivedAt < (current.game_started_at ?? '')
const results = await Promise.allSettled([ || match.receivedAt < (current.game_ended_at ?? '')) return remember(current);
(async () => { return remember({ ...current, game_started_at: match.event === 'game_start' ? match.receivedAt : current.game_started_at,
const response = await fetch('/api/servers', { signal: controller.signal }); game_ended_at: match.event === 'game_end' ? match.receivedAt : null });
if (!response.ok) throw new Error('Could not refresh servers. Retrying automatically.'); });
const servers = await response.json(); const names = new Set(snapshot.servers.map(server => server.name));
if (servers.serverNow < snapshotServerNow) return; for (const name of observed.keys()) if (!names.has(name)) observed.delete(name);
snapshotServerNow = servers.serverNow; return next;
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 }));
})()
]);
const failure = results.find(result => result.status === 'rejected');
if (failure?.status === 'rejected') throw failure.reason;
fleetError.set('');
} catch (error) {
fleetError.set(error instanceof Error && error.name !== 'AbortError'
? error.message : 'Fleet refresh timed out. Retrying automatically.');
} finally {
clearTimeout(timeout);
fleetRefreshing.set(false);
}
})().finally(() => {
inFlight = null;
if (refreshRequested) { refreshRequested = false; void refreshFleet(); }
}); });
return inFlight;
let snapshotServerNow = 0;
function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; 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<void> | null = null;
let refreshRequested = false;
function refreshFleet(afterCurrent = false): Promise<void> {
if (!browser) return Promise.resolve();
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 results = await Promise.allSettled([
(async () => {
const response = await fetch(serverName === undefined ? '/api/servers' : `/api/servers/${encodeURIComponent(serverName)}`, { 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,
operations: serverName === undefined ? value.operations : servers.operations }));
})(),
(async () => {
if (serverName !== undefined) return;
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 }));
})()
]);
const failure = results.find(result => result.status === 'rejected');
if (failure?.status === 'rejected') throw failure.reason;
fleetError.set('');
} catch (error) {
fleetError.set(error instanceof Error && error.name !== 'AbortError'
? error.message : 'Fleet refresh timed out. Retrying automatically.');
} finally {
clearTimeout(timeout);
fleetRefreshing.set(false);
}
})().finally(() => {
inFlight = null;
if (refreshRequested) { refreshRequested = false; void refreshFleet(); }
});
return inFlight;
}
let stop: (() => void) | null = null;
function startFleetMonitor(): () => void {
if (!browser || stop) return () => {};
const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000);
let refreshTimer: ReturnType<typeof setTimeout> | undefined;
let stopped = false;
let pending = false;
let refreshing = false;
let lastEventRefresh = -Infinity;
const schedule = () => {
if (stopped || refreshing || refreshTimer !== undefined || !pending || document.visibilityState === 'hidden') return;
refreshTimer = setTimeout(async () => {
refreshTimer = undefined;
refreshing = true;
try {
// Wait for any manual/poll refresh, then fetch evidence newer than
// the event. Do not use refreshFleet(true)'s immediate follow-up.
if (inFlight) await inFlight;
if (stopped || document.visibilityState === 'hidden') return;
pending = false;
lastEventRefresh = Date.now();
await refreshFleet();
} finally {
refreshing = false;
schedule();
}
}, Math.max(250, lastEventRefresh + 2000 - Date.now()));
};
const changed = () => {
if (document.visibilityState === 'hidden') return;
pending = true;
schedule();
};
window.addEventListener('ec:fleet-changed', changed);
stop = () => {
stopped = true;
stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null;
};
return stop;
}
return { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor };
} }
let stop: (() => void) | null = null; export const { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor } = createFleetStore();
export function startFleetMonitor(): () => void {
if (!browser || stop) return () => {};
const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000);
let refreshTimer: ReturnType<typeof setTimeout> | undefined;
const changed = () => {
if (refreshTimer || document.visibilityState === 'hidden') return;
refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250);
};
window.addEventListener('ec:fleet-changed', changed);
stop = () => {
stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null;
};
return stop;
}
+61 -25
View File
@@ -1,4 +1,4 @@
import { writable, readonly } from 'svelte/store'; import { writable, readonly, get } from 'svelte/store';
import { browser } from '$app/environment'; import { browser } from '$app/environment';
import type { GameTelemetry } from '$lib/telemetry.js'; import type { GameTelemetry } from '$lib/telemetry.js';
@@ -56,13 +56,17 @@ export function serverTimeNow(): number {
return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now(); return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now();
} }
export function syncServerTime(serverNow?: number): void { function updateServerClock(serverNow?: number): void {
if (typeof serverNow === 'number' && Number.isFinite(serverNow)) { if (typeof serverNow === 'number' && Number.isFinite(serverNow)) {
// A delayed HTTP snapshot must not rewind time behind a newer stream // A delayed HTTP snapshot must not rewind time behind a newer stream
// update and make that update look like it came from the future. // update and make that update look like it came from the future.
const time = serverClock ? Math.max(serverNow, serverTimeNow()) : serverNow; const time = serverClock ? Math.max(serverNow, serverTimeNow()) : serverNow;
serverClock = { time, receivedAt: performance.now() }; serverClock = { time, receivedAt: performance.now() };
} }
}
export function syncServerTime(serverNow?: number): void {
updateServerClock(serverNow);
// Refresh immediately: a suspended tab's interval may be well behind. // Refresh immediately: a suspended tab's interval may be well behind.
_uiNow.set(serverTimeNow()); _uiNow.set(serverTimeNow());
} }
@@ -74,35 +78,67 @@ let source: EventSource | null = null;
export function startServerStream(): (() => void) | undefined { export function startServerStream(): (() => void) | undefined {
if (!browser || source) return; if (!browser || source) return;
syncServerTime(); syncServerTime();
source = new EventSource('/api/events/servers'); const stream = source = new EventSource('/api/events/servers');
source.addEventListener('server-telemetry', (evt) => { let stopped = false;
let frame: number | null = null;
let pendingStatus: Map<string, ServerStatus> | null = null;
let pendingTelemetry: Map<string, GameTelemetry> | null = null;
let pendingMatches: Map<string, GameTelemetry> | null = null;
let receivedAt = 0;
const flush = () => {
frame = null;
if (stopped) return;
const statuses = pendingStatus, telemetry = pendingTelemetry, matches = pendingMatches;
const statusAt = receivedAt;
pendingStatus = pendingTelemetry = pendingMatches = null;
// Publish the latest clock before exposing reports, including after a
// suspended frame. Every changed store is published at most once per frame.
syncServerTime();
if (statuses) { statusByServer.set(statuses); statusReceivedAt.set(statusAt); }
if (matches) matchEventsByServer.set(matches);
if (telemetry) telemetryByServer.set(telemetry);
};
const schedule = () => { if (frame === null) frame = requestAnimationFrame(flush); };
stream.addEventListener('server-telemetry', (evt) => {
if (stopped) return;
const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number }; const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number };
syncServerTime(data.serverNow); updateServerClock(data.serverNow);
matchEventsByServer.update((previous) => { // Merge into one private map per frame. Snapshots replace all earlier
const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous); // evidence; later deltas still apply in arrival order.
for (const entry of data.entries) { if (data.snapshot) {
if (entry.event === 'game_start' || entry.event === 'game_end') next.set(entry.server, entry); pendingTelemetry = new Map();
pendingMatches = new Map();
}
pendingTelemetry ??= new Map(get(telemetryByServer));
for (const entry of data.entries) {
pendingTelemetry.set(entry.server, entry);
if (entry.event === 'game_start' || entry.event === 'game_end') {
pendingMatches ??= new Map(get(matchEventsByServer));
pendingMatches.set(entry.server, entry);
} }
return next; }
}); schedule();
telemetryByServer.update((previous) => {
const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
for (const entry of data.entries) next.set(entry.server, entry);
return next;
});
}); });
source.addEventListener('server-status', (evt) => { stream.addEventListener('server-status', (evt) => {
if (stopped) return;
const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean }; const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean };
syncServerTime(data.serverNow); updateServerClock(data.serverNow);
statusByServer.update(previous => { receivedAt = Date.now();
const m = data.delta ? new Map(previous) : new Map<string, ServerStatus>(); if (!data.delta) pendingStatus = new Map();
for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() }); pendingStatus ??= new Map(get(statusByServer));
return m; for (const entry of data.entries) pendingStatus.set(entry.server, { ...entry, receivedAt });
}); schedule();
statusReceivedAt.set(Date.now());
}); });
if (tickInterval === null) { if (tickInterval === null) {
tickInterval = setInterval(() => { if (typeof document === 'undefined' || document.visibilityState !== 'hidden') _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; }; return () => {
if (stopped) return;
stopped = true;
if (frame !== null) cancelAnimationFrame(frame);
pendingStatus = pendingTelemetry = pendingMatches = null;
stream.close(); source = null;
if (tickInterval) clearInterval(tickInterval);
tickInterval = null;
};
} }
+1 -1
View File
@@ -75,7 +75,7 @@
$effect(() => { $effect(() => {
if (!sessionReady) return; if (!sessionReady) return;
const path = $page.url.pathname; const path = $page.url.pathname;
if (!(path === '/' || /^\/(servers|scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return; if (!(path === '/' || /^\/(scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return;
if ('fleetSnapshot' in $page.data && $page.data.fleetSnapshot) hydrateFleet($page.data.fleetSnapshot); if ('fleetSnapshot' in $page.data && $page.data.fleetSnapshot) hydrateFleet($page.data.fleetSnapshot);
return startFleetMonitor(); return startFleetMonitor();
}); });
+22 -12
View File
@@ -1,3 +1,7 @@
<script module lang="ts">
const serverTitleCollator = new Intl.Collator(undefined, { numeric: true });
</script>
<script lang="ts"> <script lang="ts">
let { data }: { data: import('./$types').PageData } = $props(); let { data }: { data: import('./$types').PageData } = $props();
const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot); const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
@@ -92,15 +96,21 @@
let problems = $derived(servers.filter(server => server.health === 'offline' || server.health === 'stale')); let problems = $derived(servers.filter(server => server.health === 'offline' || server.health === 'stale'));
let starting = $derived(servers.filter(server => server.health === 'starting').length); let starting = $derived(servers.filter(server => server.health === 'starting').length);
let provisioning = $derived(servers.filter(server => server.health === 'provisioning').length); let provisioning = $derived(servers.filter(server => server.health === 'provisioning').length);
let filtered = $derived(servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, observationNow())) let filtered = $derived.by(() => {
.sort((a, b) => { const now = observationNow();
const pinOrder = Number(pinned.includes(b.name)) - Number(pinned.includes(a.name)); const pinnedNames = new Set(pinned);
if (pinOrder) return pinOrder; return servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, now))
const rank = (server: FleetServer) => sort === 'players' ? (freshServer(server, observationNow()) ? server.last_players ?? -1 : -1) .map(server => ({
: sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), observationNow()) ?? -1 server,
: sort === 'duration' ? matchDuration(server) : 0; title: serverTitle(server.name),
return rank(b) - rank(a) || serverTitle(a.name).localeCompare(serverTitle(b.name), undefined, { numeric: true }); pinned: Number(pinnedNames.has(server.name)),
})); rank: sort === 'players' ? (freshServer(server, now) ? server.last_players ?? -1 : -1)
: sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), now) ?? -1
: sort === 'duration' ? matchDuration(server, now) : 0
}))
.sort((a, b) => b.pinned - a.pinned || b.rank - a.rank || serverTitleCollator.compare(a.title, b.title))
.map(({ server }) => server);
});
let groups = $derived.by(() => { let groups = $derived.by(() => {
const result = new Map<string, { title: string; game: string; servers: FleetServer[]; launch: ReturnType<typeof fleetIdentity>['launch'] }>(); const result = new Map<string, { title: string; game: string; servers: FleetServer[]; launch: ReturnType<typeof fleetIdentity>['launch'] }>();
for (const server of filtered) { for (const server of filtered) {
@@ -116,9 +126,9 @@
let spotlight = $derived(servers.filter(server => currentRound(server, $telemetryByServer.get(server.name), observationNow()) !== null) let spotlight = $derived(servers.filter(server => currentRound(server, $telemetryByServer.get(server.name), observationNow()) !== null)
.sort((a, b) => matchDuration(b) - matchDuration(a))[0]); .sort((a, b) => matchDuration(b) - matchDuration(a))[0]);
function matchDuration(server: FleetServer): number { function matchDuration(server: FleetServer, now = observationNow()): number {
return freshServer(server, observationNow()) && server.active_since && server.game_started_at && !server.game_ended_at return freshServer(server, now) && server.active_since && server.game_started_at && !server.game_ended_at
&& (server.last_players ?? 0) > 0 ? Math.max(0, observationNow() - Date.parse(server.game_started_at)) : -1; && (server.last_players ?? 0) > 0 ? Math.max(0, now - Date.parse(server.game_started_at)) : -1;
} }
function togglePin(name: string) { pinned = pinned.includes(name) ? pinned.filter(item => item !== name) : [...pinned, name]; } function togglePin(name: string) { pinned = pinned.includes(name) ? pinned.filter(item => item !== name) : [...pinned, name]; }
function launch(selection: { game: ZombieGame; slug: string } | null = null) { function launch(selection: { game: ZombieGame; slug: string } | null = null) {
+1 -2
View File
@@ -2,7 +2,7 @@ import { json } from '@sveltejs/kit';
import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index.js'; import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index.js';
import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js';
import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js';
import { reconcileHostLibraryCleanup, cleanupControllerLibraries } from '$lib/server/game-files/index.js'; import { reconcileHostLibraryCleanup } from '$lib/server/game-files/index.js';
import { fleetSettings } from '$lib/server/settings.js'; import { fleetSettings } from '$lib/server/settings.js';
export async function POST({ request }: { request: Request }) { export async function POST({ request }: { request: Request }) {
const authenticated = authenticateAgent(request); const authenticated = authenticateAgent(request);
@@ -16,7 +16,6 @@ export async function POST({ request }: { request: Request }) {
} }
reconcileHostPreparation(host.id); reconcileHostPreparation(host.id);
reconcileHostLibraryCleanup(host.id); reconcileHostLibraryCleanup(host.id);
await cleanupControllerLibraries();
return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15, return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15,
observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() }); observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() });
} catch (cause) { return hostApiError(cause); } } catch (cause) { return hostApiError(cause); }
+10 -2
View File
@@ -1,11 +1,14 @@
import { observationReader } from '$lib/server/fleet/status.js'; import { observationReader } from '$lib/server/fleet/status.js';
import { listServers } from '$lib/server/fleet/servers.js'; import { listServers } from '$lib/server/fleet/servers.js';
import { statusEntry } from '$lib/server/pipeline/worker.js'; import { statusEntry } from '$lib/server/pipeline/worker.js';
import { subscribe } from '$lib/server/pipeline/bus.js'; import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js';
import { listTelemetry } from '$lib/server/fleet/telemetry.js'; import { listTelemetry } from '$lib/server/fleet/telemetry.js';
const encoder = new TextEncoder(); const encoder = new TextEncoder();
const HEARTBEAT_MS = 20_000; const HEARTBEAT_MS = 20_000;
// Bus subscribers receive the same event object. Reuse its complete frame
// across connections without retaining events after delivery.
const frames = new WeakMap<PipelineEvent, Uint8Array>();
export async function GET() { export async function GET() {
let heartbeat: NodeJS.Timeout | null = null; let heartbeat: NodeJS.Timeout | null = null;
@@ -15,7 +18,12 @@ export async function GET() {
start(controller) { start(controller) {
unsubscribe = subscribe((event) => { unsubscribe = subscribe((event) => {
if (event.type === 'server-status' || event.type === 'server-telemetry') { if (event.type === 'server-status' || event.type === 'server-telemetry') {
controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`)); let frame = frames.get(event);
if (!frame) {
frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
frames.set(event, frame);
}
controller.enqueue(frame);
} }
}); });
const observation = observationReader(); const observation = observationReader();
+5
View File
@@ -1,6 +1,11 @@
import { kick } from '$lib/server/pipeline/worker.js'; import { kick } from '$lib/server/pipeline/worker.js';
import { json } from '@sveltejs/kit'; import { json } from '@sveltejs/kit';
import { destroyServer } from '$lib/server/provisioning/index.js'; import { destroyServer } from '$lib/server/provisioning/index.js';
import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
export function GET({ params }: { params: { name: string } }) {
return json(serverSnapshot(params.name));
}
export async function DELETE({ params }: { params: { name: string } }) { export async function DELETE({ params }: { params: { name: string } }) {
const name = params.name; const name = params.name;
+4 -3
View File
@@ -1,4 +1,5 @@
import { fleetSnapshot } from '$lib/server/fleet/snapshot.js'; import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
import { fleetOperations } from '$lib/server/fleet/overview.js';
export function load() { return { fleetSnapshot: { ...fleetSnapshot(), operations: fleetOperations() } }; } export function load({ params }: { params: { name: string } }) {
return { serverSnapshot: serverSnapshot(params.name) };
}
+11 -5
View File
@@ -1,14 +1,12 @@
<script lang="ts"> <script lang="ts">
let { data }: { data: import('./$types').PageData } = $props(); let { data }: { data: import('./$types').PageData } = $props();
const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.fleetSnapshot.servers);
import RuntimeVersion from '$ui/components/RuntimeVersion.svelte'; import RuntimeVersion from '$ui/components/RuntimeVersion.svelte';
import MatchHistory from '$ui/components/MatchHistory.svelte'; import MatchHistory from '$ui/components/MatchHistory.svelte';
import RoundTelemetry from '$ui/components/RoundTelemetry.svelte'; import RoundTelemetry from '$ui/components/RoundTelemetry.svelte';
import TelemetryDiagnostics from '$ui/components/TelemetryDiagnostics.svelte'; import TelemetryDiagnostics from '$ui/components/TelemetryDiagnostics.svelte';
import { pollVisible } from '$ui/poll.js'; import { pollVisible } from '$ui/poll.js';
import { fetchCampaigns } from '$ui/campaigns.js'; import { fetchCampaigns } from '$ui/campaigns.js';
import { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } from '$ui/stores/fleet.js'; import { createFleetStore } from '$ui/stores/fleet.js';
import { freshServer, fleetIdentity } from '$lib/fleet-overview.js'; import { freshServer, fleetIdentity } from '$lib/fleet-overview.js';
import { serverTitle } from '$ui/server-labels.js'; import { serverTitle } from '$ui/server-labels.js';
import ServerIdentity from '$ui/components/ServerIdentity.svelte'; import ServerIdentity from '$ui/components/ServerIdentity.svelte';
@@ -109,9 +107,17 @@
} }
let name = $derived($page.params.name ?? ''); let name = $derived($page.params.name ?? '');
const detail = $derived(createFleetStore(name));
const { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } = $derived(detail);
$effect(() => {
detail.hydrateFleet(data.serverSnapshot);
return detail.startFleetMonitor();
});
const snapshot = $derived($fleet.receivedAt ? $fleet : data.serverSnapshot);
const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.serverSnapshot.servers);
let server = $derived((visibleFleet.find(s => s.name === name) as Server | undefined) ?? null); let server = $derived((visibleFleet.find(s => s.name === name) as Server | undefined) ?? null);
const native = $derived(!!server?.native_instance_id); const native = $derived(!!server?.native_instance_id);
let integrations = $derived($fleetIntegrations ?? data.fleetSnapshot.integrations); let integrations = $derived($fleetIntegrations ?? data.serverSnapshot.integrations);
let history = $state<DeploymentRow[]>([]); let history = $state<DeploymentRow[]>([]);
let inventory = $state<InventoryEntry[]>([]); let inventory = $state<InventoryEntry[]>([]);
let komodoProbe = $state<KomodoProbeResult | null>(null); let komodoProbe = $state<KomodoProbeResult | null>(null);
@@ -610,7 +616,7 @@
</div> </div>
{:else if $fleetError} {:else if $fleetError}
<div class="ui-error" role="alert">{$fleetError}<Button onclick={() => refreshFleet(true)}>Retry</Button></div> <div class="ui-error" role="alert">{$fleetError}<Button onclick={() => refreshFleet(true)}>Retry</Button></div>
{:else if $fleet.receivedAt || data.fleetSnapshot} {:else if $fleet.receivedAt || data.serverSnapshot}
<EmptyState title="Server not found" description="This server may have been removed from the fleet." /> <EmptyState title="Server not found" description="This server may have been removed from the fleet." />
{:else} {:else}
<div class="loading">Loading…</div> <div class="loading">Loading…</div>
+31 -5
View File
@@ -32,7 +32,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
VERSION = "1.4.10" VERSION = "1.4.11"
LABEL = "io.electric-cherry" LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024 MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000 MANIFEST_MAX_ENTRIES = 10000
@@ -918,6 +918,8 @@ class Agent:
self.observation_keep_days = 0 # Older controllers do not authorize cleanup. self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
self.library_progress = {} self.library_progress = {}
self.library_transfers = {} self.library_transfers = {}
self.inventory_manifest_cache = {}
self.inventory_manifest_lock = threading.Lock()
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json") self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
self.executing_id = None self.executing_id = None
@@ -1191,6 +1193,25 @@ class Agent:
continue continue
return samples return samples
def inventory_manifest(self, marker):
# Inventory only: launch/download verification must still read disk.
# Include file identity so atomic replacements invalidate even when
# their size and modification time match the previous manifest.
with self.inventory_manifest_lock:
try:
metadata = marker.stat()
signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
cached = self.inventory_manifest_cache.get(marker)
if cached is not None and cached[0] == signature:
return cached[1]
self.inventory_manifest_cache.pop(marker, None)
value = json.loads(marker.read_text())
self.inventory_manifest_cache[marker] = (signature, value)
return value
except (OSError, ValueError):
self.inventory_manifest_cache.pop(marker, None)
raise
def inventory(self): def inventory(self):
cpu_usage = self.cpu_usage_percent() cpu_usage = self.cpu_usage_percent()
usage = shutil.disk_usage(self.root) usage = shutil.disk_usage(self.root)
@@ -1210,13 +1231,18 @@ class Agent:
libraries, caches = [], [] libraries, caches = [], []
progress = {key: value.copy() for key, value in list(self.library_progress.items())} progress = {key: value.copy() for key, value in list(self.library_progress.items())}
progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())}) progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
for marker in (self.root / "libraries").glob("*/*/manifest.json"): manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
with self.inventory_manifest_lock:
for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
del self.inventory_manifest_cache[marker]
for marker in manifests:
# A crash between writing a manifest and its atomic promotion must # A crash between writing a manifest and its atomic promotion must
# never expose the staging directory as a ready library. # never expose the staging directory as a ready library.
if marker.parent.name.startswith("."): if marker.parent.name.startswith("."):
continue continue
try: try:
value = json.loads(marker.read_text()) value = self.inventory_manifest(marker)
# Rechecking an existing cache must revoke readiness until its # Rechecking an existing cache must revoke readiness until its
# entire contents have passed verification again. # entire contents have passed verification again.
if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists(): if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
@@ -1233,9 +1259,9 @@ class Agent:
except (OSError, ValueError, KeyError): except (OSError, ValueError, KeyError):
continue continue
ready = {item["id"] for item in libraries} ready = {item["id"] for item in libraries}
for marker in (self.root / "libraries").glob("*/.*.partial/source.json"): for marker in partial_manifests:
try: try:
value = json.loads(marker.read_text()) value = self.inventory_manifest(marker)
library_id = identifier(value["libraryId"]) library_id = identifier(value["libraryId"])
if library_id in ready or library_id in progress: if library_id in ready or library_id in progress:
continue continue
+2 -2
View File
@@ -1,8 +1,8 @@
{ {
"version": "1.4.10", "version": "1.4.11",
"protocolVersion": 1, "protocolVersion": 1,
"platform": "debian-amd64", "platform": "debian-amd64",
"sha256": "c518a216ad4e9dd86ac82743b27bae54cdf1c51ef7ae3bf8d252dcc9863be1fb", "sha256": "4c6cc180f3c45a436878a0278d62b6840526150e039fdf51bafadef63759b131",
"updaterProtocolVersion": 1, "updaterProtocolVersion": 1,
"updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
} }
+5 -1
View File
@@ -48,7 +48,11 @@ beforeEach(async () => {
vi.mocked(komodo.komodoUpdateTagsOnResource).mockResolvedValue({ ok: true }); vi.mocked(komodo.komodoUpdateTagsOnResource).mockResolvedValue({ ok: true });
const rcon = await import('../../src/lib/server/integrations/rcon.js'); const rcon = await import('../../src/lib/server/integrations/rcon.js');
vi.mocked(rcon.rconGetPlayerCount).mockResolvedValue(0); vi.mocked(rcon.rconGetPlayerCount).mockResolvedValue(0);
vi.mocked(rcon.rconPlutoniumVersion).mockResolvedValue({ revision: 5354, error: null }); vi.mocked(rcon.rconPlutoniumVersion).mockImplementation(async () => {
// Version evidence must be observed strictly after the launch timestamp.
await new Promise(resolve => setTimeout(resolve, 2));
return { revision: 5354, error: null };
});
vi.mocked(rcon.rconStatusResult).mockImplementation(async server => status(server)); vi.mocked(rcon.rconStatusResult).mockImplementation(async server => status(server));
const { kick } = await import('../../src/lib/server/pipeline/worker.js'); const { kick } = await import('../../src/lib/server/pipeline/worker.js');
vi.mocked(kick).mockResolvedValue(undefined); vi.mocked(kick).mockResolvedValue(undefined);
+45
View File
@@ -0,0 +1,45 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { newFixture, type Fixture } from '../helpers/fixture.js';
let fixture: Fixture;
beforeEach(async () => {
vi.resetModules();
fixture = await newFixture([]);
});
afterEach(async () => { await fixture.cleanup(); });
describe('prepared statement caching', () => {
it('reuses statements with fresh parameters and remains usable after rollback', async () => {
const { db } = await import('../../src/lib/server/db/index.js');
const insertSql = 'INSERT INTO settings (key, value) VALUES (?, ?)';
const selectSql = 'SELECT value FROM settings WHERE key = ?';
const insert = db().prepare(insertSql);
const select = db().prepare(selectSql);
insert.run('cache-first', 'one');
expect(() => db().transaction(() => {
db().prepare(insertSql).run('cache-rollback', 'discarded');
throw new Error('rollback');
})()).toThrow('rollback');
db().prepare(insertSql).run('cache-second', 'two');
expect(db().prepare(insertSql)).toBe(insert);
expect(db().prepare(selectSql)).toBe(select);
expect(select.get('cache-first')).toEqual({ value: 'one' });
expect(select.get('cache-second')).toEqual({ value: 'two' });
expect(select.get('cache-rollback')).toBeUndefined();
});
it('prepares fresh statements when the connection is reopened', async () => {
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
const sql = 'SELECT value FROM settings WHERE key = ?';
const oldStatement = db().prepare(sql);
db().prepare('INSERT INTO settings (key, value) VALUES (?, ?)').run('cache-persisted', 'saved');
closeDb();
const statement = db().prepare(sql);
expect(statement).not.toBe(oldStatement);
expect(statement.database).not.toBe(oldStatement.database);
expect(statement.get('cache-persisted')).toEqual({ value: 'saved' });
expect(db().prepare(sql)).toBe(statement);
});
});
+5 -1
View File
@@ -15,7 +15,11 @@ vi.mock('../../src/lib/server/integrations/komodo.js', () => ({
})); }));
vi.mock('../../src/lib/server/integrations/rcon.js', () => ({ vi.mock('../../src/lib/server/integrations/rcon.js', () => ({
rconGetPlayerCount: vi.fn(async () => 0), rconGetPlayerCount: vi.fn(async () => 0),
rconPlutoniumVersion: vi.fn(async () => ({ revision: 5354, error: null })), rconPlutoniumVersion: vi.fn(async () => {
// Version evidence must be observed strictly after the launch timestamp.
await new Promise(resolve => setTimeout(resolve, 2));
return { revision: 5354, error: null };
}),
rconStatusResult: vi.fn(async (server: string) => ({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: '', rconStatusResult: vi.fn(async (server: string) => ({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: '',
status: { players: 0, map: server.startsWith('t6_') ? 'zm_transit' : server.includes('_moon_') ? 'zombie_moon' : 'zombie_theater', gametype: null, roster: [] } })) status: { players: 0, map: server.startsWith('t6_') ? 'zm_transit' : server.includes('_moon_') ? 'zombie_moon' : 'zombie_theater', gametype: null, roster: [] } }))
})); }));
+158
View File
@@ -0,0 +1,158 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { get } from 'svelte/store';
import type { FleetOperation } from '../../src/lib/fleet-overview.js';
vi.mock('$app/environment', () => ({ browser: true }));
let documentStub: EventTarget & { visibilityState: string };
let stop: (() => void) | undefined;
const changed = () => window.dispatchEvent(new Event('ec:fleet-changed'));
const visibility = (state: string) => {
documentStub.visibilityState = state;
documentStub.dispatchEvent(new Event('visibilitychange'));
};
const response = (url: string) => Response.json(url === '/api/servers'
? { serverNow: Date.now(), servers: [], integrations: {} } : { operations: [] });
beforeEach(() => {
vi.resetModules();
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-10-05T12:00:00Z'));
documentStub = Object.assign(new EventTarget(), { visibilityState: 'visible' });
vi.stubGlobal('document', documentStub);
vi.stubGlobal('window', new EventTarget());
vi.stubGlobal('fetch', vi.fn(async (url: string) => response(url)));
});
afterEach(() => {
stop?.(); stop = undefined;
vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks();
});
async function setup() {
const store = await import('../../src/lib/ui/stores/fleet.js');
store.fleet.set({ servers: [], operations: [], receivedAt: Date.now() });
stop = store.startFleetMonitor();
return store;
}
describe('fleet event refreshes', () => {
it('refreshes server history through one scoped request without replacing the fleet snapshot', async () => {
const store = await import('../../src/lib/ui/stores/fleet.js');
const integrations = { rcon: true, komodo: true, rconTimeoutMs: 2000, defaultRconHost: '' };
const snapshot = { servers: [], operations: [], integrations, serverNow: Date.now() };
store.hydrateFleet(snapshot);
const originalFleet = get(store.fleet);
const detail = store.createFleetStore('one / two');
detail.hydrateFleet(snapshot);
stop = detail.startFleetMonitor();
const history: FleetOperation[] = [{ key: 'deployment:done', group: 'deployment:done', kind: 'deployment',
server: 'one / two', state: 'done', label: 'Complete', detail: 'Complete', href: '/provisioning/done', startedAt: new Date().toISOString() }];
vi.mocked(fetch).mockResolvedValue(Response.json({ ...snapshot, operations: history }));
changed();
await vi.advanceTimersByTimeAsync(250);
expect(fetch).toHaveBeenCalledTimes(1);
expect(fetch).toHaveBeenCalledWith('/api/servers/one%20%2F%20two', expect.anything());
expect(get(detail.fleet).operations).toEqual(history);
expect(get(store.fleet)).toBe(originalFleet);
expect(get(store.fleet).operations).toEqual([]);
});
it('isolates late responses when navigating between server detail stores', async () => {
const { createFleetStore, fleet } = await import('../../src/lib/ui/stores/fleet.js');
const first = createFleetStore('first');
const second = createFleetStore('second');
let finish!: (response: Response) => void;
vi.mocked(fetch).mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
const oldRequest = first.refreshFleet();
const newSnapshot = { serverNow: Date.now(), servers: [], integrations: {}, operations: [] };
vi.mocked(fetch).mockResolvedValue(Response.json(newSnapshot));
await second.refreshFleet();
const current = get(second.fleet);
finish(Response.json(newSnapshot)); await oldRequest;
expect(get(second.fleet)).toBe(current);
expect(get(fleet).receivedAt).toBe(0);
expect(fetch).toHaveBeenCalledTimes(2);
});
it('coalesces bursts and refreshes at most once every two seconds with a trailing refresh', async () => {
await setup();
changed(); changed(); changed();
await vi.advanceTimersByTimeAsync(249);
expect(fetch).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(fetch).toHaveBeenCalledTimes(2);
for (let i = 0; i < 7; i++) {
changed();
await vi.advanceTimersByTimeAsync(250);
expect(fetch).toHaveBeenCalledTimes(2);
}
await vi.advanceTimersByTimeAsync(250);
expect(fetch).toHaveBeenCalledTimes(4);
changed();
await vi.advanceTimersByTimeAsync(2000);
expect(fetch).toHaveBeenCalledTimes(6);
await vi.advanceTimersByTimeAsync(5000);
expect(fetch).toHaveBeenCalledTimes(6);
});
it('keeps events during a request without an immediate follow-up when it finishes', async () => {
await setup();
let finish!: () => void;
const gate = new Promise<void>(resolve => { finish = resolve; });
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
changed();
await vi.advanceTimersByTimeAsync(250);
expect(fetch).toHaveBeenCalledTimes(2);
changed(); changed();
await vi.advanceTimersByTimeAsync(500);
finish();
await vi.advanceTimersByTimeAsync(1499);
expect(fetch).toHaveBeenCalledTimes(2);
await vi.advanceTimersByTimeAsync(1);
expect(fetch).toHaveBeenCalledTimes(4);
});
it('waits for a slow manual refresh and coalesces events arriving during it', async () => {
const { refreshFleet } = await setup();
let finish!: () => void;
const gate = new Promise<void>(resolve => { finish = resolve; });
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
const manual = refreshFleet();
changed();
await vi.advanceTimersByTimeAsync(3000);
changed(); changed();
expect(fetch).toHaveBeenCalledTimes(2);
finish(); await manual;
await vi.advanceTimersByTimeAsync(0);
expect(fetch).toHaveBeenCalledTimes(4);
await vi.advanceTimersByTimeAsync(3000);
expect(fetch).toHaveBeenCalledTimes(4);
});
it('skips queued refreshes while hidden, refreshes on return, and cancels on stop', async () => {
await setup();
changed(); visibility('hidden');
await vi.advanceTimersByTimeAsync(250);
changed();
await vi.advanceTimersByTimeAsync(3000);
expect(fetch).not.toHaveBeenCalled();
visibility('visible');
await vi.advanceTimersByTimeAsync(0);
expect(fetch).toHaveBeenCalledTimes(2);
changed(); stop!();
await vi.advanceTimersByTimeAsync(3000);
expect(fetch).toHaveBeenCalledTimes(2);
});
it('cancels a follow-up waiting for an active request when stopped', async () => {
const { refreshFleet } = await setup();
let finish!: () => void;
const gate = new Promise<void>(resolve => { finish = resolve; });
vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
const manual = refreshFleet();
changed();
await vi.advanceTimersByTimeAsync(250);
stop!(); finish(); await manual;
await vi.advanceTimersByTimeAsync(3000);
expect(fetch).toHaveBeenCalledTimes(2);
});
});
+17 -12
View File
@@ -64,8 +64,10 @@ describe('persisted Fleet operations', () => {
const { db } = await import('../../src/lib/server/db/index.js'); const { db } = await import('../../src/lib/server/db/index.js');
const handle = db(); const handle = db();
handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at,detail) VALUES ('server','update','failed',5354,?,'Revision not verified')").run(iso()); handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at,detail) VALUES ('server','update','failed',5354,?,'Revision not verified')").run(iso());
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js'); const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
expect(fleetOperations(now)).toEqual([expect.objectContaining({ kind: 'maintenance', state: 'attention', detail: 'Revision not verified' })]); expect(fleetOperations(now)).toEqual([expect.objectContaining({ kind: 'maintenance', state: 'attention', detail: 'Revision not verified' })]);
expect(serverOperations('server', now)).toEqual(fleetOperations(now));
expect(serverOperations('unrelated', now)).toEqual([]);
handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at) VALUES ('server','update','queued',5354,?)").run(iso(1000)); handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at) VALUES ('server','update','queued',5354,?)").run(iso(1000));
expect(fleetOperations(now)).toEqual([]); expect(fleetOperations(now)).toEqual([]);
}); });
@@ -76,7 +78,7 @@ describe('persisted Fleet operations', () => {
handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run(); handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
handle.prepare("INSERT INTO deployments (id, action, started_at) VALUES ('one', 'promote', ?)").run(iso(-2 * 86_400_000)); handle.prepare("INSERT INTO deployments (id, action, started_at) VALUES ('one', 'promote', ?)").run(iso(-2 * 86_400_000));
handle.prepare("INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state, players) VALUES ('one', 'server', 1, 'pending', 4)").run(); handle.prepare("INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state, players) VALUES ('one', 'server', 1, 'pending', 4)").run();
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js'); const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
expect(fleetOperations(now)).toEqual([expect.objectContaining({ label: 'Waiting for an empty lobby', state: 'active', href: '/campaigns/one' })]); expect(fleetOperations(now)).toEqual([expect.objectContaining({ label: 'Waiting for an empty lobby', state: 'active', href: '/campaigns/one' })]);
handle.prepare("UPDATE deployment_servers SET restarted_at = ?, players = NULL").run(iso()); handle.prepare("UPDATE deployment_servers SET restarted_at = ?, players = NULL").run(iso());
handle.prepare("INSERT INTO restart_rollout (server, requested_at) VALUES ('server', ?)").run(iso()); handle.prepare("INSERT INTO restart_rollout (server, requested_at) VALUES ('server', ?)").run(iso());
@@ -86,18 +88,20 @@ describe('persisted Fleet operations', () => {
expect(fleetOperations(now)[0].label).toBe('Recovering'); expect(fleetOperations(now)[0].label).toBe('Recovering');
db().prepare("UPDATE deployment_servers SET restart_state = 'done', checked_at = ?").run(iso()); db().prepare("UPDATE deployment_servers SET restart_state = 'done', checked_at = ?").run(iso());
db().prepare('UPDATE restart_rollout SET healthy_at = ?').run(iso()); db().prepare('UPDATE restart_rollout SET healthy_at = ?').run(iso());
expect(fleetOperations(now)[0].state).toBe('done'); expect(fleetOperations(now)).toEqual([]);
expect(serverOperations('server', now)[0].state).toBe('done');
}); });
it('keeps lifecycle failures actionable until superseded and retains successful completions', async () => { it('keeps lifecycle failures actionable until superseded and retains completions only for that server', async () => {
const { db } = await import('../../src/lib/server/db/index.js'); const { db } = await import('../../src/lib/server/db/index.js');
const handle = db(); const handle = db();
const insert = handle.prepare("INSERT INTO server_deployments (id, server, action, state, stage, started_at, finished_at, triggered_by) VALUES (?, ?, ?, ?, ?, ?, ?, 'test')"); const insert = handle.prepare("INSERT INTO server_deployments (id, server, action, state, stage, started_at, finished_at, triggered_by) VALUES (?, ?, ?, ?, ?, ?, ?, 'test')");
insert.run('failed', 'server', 'create', 'failed', 'Creating container', iso(-86_400_001), iso(-86_400_001)); insert.run('failed', 'server', 'create', 'failed', 'Creating container', iso(-86_400_001), iso(-86_400_001));
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js'); const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
expect(fleetOperations(now)[0].state).toBe('attention'); expect(fleetOperations(now)[0].state).toBe('attention');
insert.run('retry', 'server', 'create', 'done', 'Ready', iso(), iso()); insert.run('retry', 'server', 'create', 'done', 'Ready', iso(), iso());
expect(fleetOperations(now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]); expect(fleetOperations(now)).toEqual([]);
expect(serverOperations('server', now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]);
insert.run('queued', 'new-server', 'create', 'queued', 'Waiting to launch', iso(), null); insert.run('queued', 'new-server', 'create', 'queued', 'Waiting to launch', iso(), null);
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ server: 'new-server', label: 'Waiting to launch', state: 'active' })); expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ server: 'new-server', label: 'Waiting to launch', state: 'active' }));
}); });
@@ -107,7 +111,7 @@ describe('persisted Fleet operations', () => {
const handle = db(); const handle = db();
handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run(); handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
handle.prepare("INSERT INTO restart_rollout (server, requested_at, error) VALUES ('server', ?, 'RCON not answering')").run(iso()); handle.prepare("INSERT INTO restart_rollout (server, requested_at, error) VALUES ('server', ?, 'RCON not answering')").run(iso());
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js'); const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
expect(fleetOperations(now)[0]).toMatchObject({ kind: 'recovery', state: 'attention', detail: 'RCON not answering' }); expect(fleetOperations(now)[0]).toMatchObject({ kind: 'recovery', state: 'attention', detail: 'RCON not answering' });
handle.prepare("INSERT INTO scripts (id, name, target_rel_path) VALUES (1, 'test', 'test.gsc')").run(); handle.prepare("INSERT INTO scripts (id, name, target_rel_path) VALUES (1, 'test', 'test.gsc')").run();
handle.prepare("INSERT INTO script_versions (id, script_id, version, sha256, size, content) VALUES (1, 1, 1, 'hash', 0, X'')").run(); handle.prepare("INSERT INTO script_versions (id, script_id, version, sha256, size, content) VALUES (1, 1, 1, 'hash', 0, X'')").run();
@@ -115,9 +119,11 @@ describe('persisted Fleet operations', () => {
handle.prepare("INSERT INTO beta_servers (beta_id, server, had_original, stage) VALUES (1, 'server', 0, 'awaiting-match')").run(); handle.prepare("INSERT INTO beta_servers (beta_id, server, had_original, stage) VALUES (1, 'server', 0, 'awaiting-match')").run();
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'beta', label: 'Observing a match', href: '/betas/1' })); expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'beta', label: 'Observing a match', href: '/betas/1' }));
expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'recovery', state: 'attention' })); expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'recovery', state: 'attention' }));
expect(serverOperations('server', now)).toEqual(fleetOperations(now));
expect(serverOperations('unrelated', now)).toEqual([]);
}); });
it('preserves lifecycle selection across cutoff, deletion, supersession, and overlapping warning branches', async () => { it('keeps only active work and unresolved failures across deletion and supersession', async () => {
const { db } = await import('../../src/lib/server/db/index.js'); const { db } = await import('../../src/lib/server/db/index.js');
const handle = db(); const handle = db();
const since = iso(-86_400_000); const since = iso(-86_400_000);
@@ -138,13 +144,12 @@ describe('persisted Fleet operations', () => {
} }
} }
})(); })();
// The former OR query is the compatibility reference for all three branches. // Completed work, including warnings and cancellations, belongs to server history.
const expected = handle.prepare(`SELECT d.id FROM server_deployments d const expected = handle.prepare(`SELECT d.id FROM server_deployments d
WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health') WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health')
OR (d.state IN ('done', 'warning', 'cancelled') AND d.finished_at >= ?) OR (d.state IN ('failed', 'interrupted') AND NOT EXISTS
OR (d.state IN ('failed', 'interrupted', 'warning') AND NOT EXISTS
(SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`) (SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`)
.all(since) as { id: string }[]; .all() as { id: string }[];
const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js'); const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
expect(fleetOperations(now).map(row => row.key).sort()).toEqual(expected.map(row => `deployment:${row.id}`).sort()); expect(fleetOperations(now).map(row => row.key).sort()).toEqual(expected.map(row => `deployment:${row.id}`).sort());
}); });
+85
View File
@@ -0,0 +1,85 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { newFixture, type Fixture } from '../helpers/fixture.js';
let fixture: Fixture;
const now = Date.parse('2026-10-05T12:00:00Z');
const iso = (offset = 0) => new Date(now + offset).toISOString();
beforeEach(async () => {
vi.resetModules();
vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now);
fixture = await newFixture([]);
const { db } = await import('../../src/lib/server/db/index.js');
db().transaction(() => {
const server = db().prepare('INSERT INTO servers(name, komodo_name) VALUES (?, ?)');
const operation = db().prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,finished_at,triggered_by)
VALUES(?,?,'restart',?,'Complete',?,?,'test')`);
for (const name of ['one', 'two']) {
server.run(name, name);
for (let i = 0; i < 100; i++) operation.run(`${name}-${i}`, name, 'done', iso(-i), iso(-i));
operation.run(`${name}-old`, name, 'done', iso(-86_400_001), iso(-86_400_001));
operation.run(`${name}-failed`, name, 'failed', iso(), iso());
operation.run(`${name}-active`, name, 'waiting', iso(), null);
}
db().prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES('campaign','push',?,?)").run(iso(), iso());
for (const name of ['one', 'two']) {
db().prepare(`INSERT INTO deployment_servers(deployment_id,server,files_changed,restart_state,checked_at)
VALUES('campaign',?,1,'done',?)`).run(name, iso());
}
})();
});
afterEach(async () => { await fixture.cleanup(); vi.useRealTimers(); vi.restoreAllMocks(); });
describe('fleet and server response scope', () => {
it('omits completed history from fleet REST and SSR responses', async () => {
const { GET } = await import('../../src/routes/api/fleet/+server.js');
const { load } = await import('../../src/routes/+page.server.js');
const body = await GET().json();
expect(body.operations.map((operation: { key: string }) => operation.key).sort())
.toEqual(['deployment:one-active', 'deployment:two-active']);
expect(load().fleetSnapshot.operations).toEqual(body.operations);
});
it('loads only the selected server and its recent history in both REST and SSR', async () => {
const { GET } = await import('../../src/routes/api/servers/[name]/+server.js');
const { load } = await import('../../src/routes/servers/[name]/+page.server.js');
const registry = await import('../../src/lib/server/fleet/servers.js');
const { fleetSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
const expected = fleetSnapshot().servers.find(server => server.name === 'one');
const list = vi.spyOn(registry, 'listServers');
const snapshot = load({ params: { name: 'one' } }).serverSnapshot;
expect(list).not.toHaveBeenCalled();
expect(snapshot.servers).toEqual([expected]);
expect(snapshot.operations).toHaveLength(103);
expect(snapshot.operations.every(operation => operation.server === 'one')).toBe(true);
expect(snapshot.operations.map(operation => operation.key)).not.toContain('deployment:one-old');
expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'campaign:campaign:one', state: 'done' }));
expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'deployment:one-failed', state: 'attention' }));
const body = await GET({ params: { name: 'one' } }).json();
expect(body).toEqual(JSON.parse(JSON.stringify(snapshot)));
expect(list).not.toHaveBeenCalled();
});
it('returns no unrelated servers or history for missing and removed servers', async () => {
const { serverSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
expect(serverSnapshot('missing')).toMatchObject({ servers: [], operations: [] });
const { db } = await import('../../src/lib/server/db/index.js');
db().prepare("UPDATE servers SET destroyed_at = ? WHERE name = 'one'").run(iso());
expect(serverSnapshot('one').servers).toEqual([]);
db().prepare("UPDATE server_deployments SET server_deleted_at = ? WHERE server = 'one'").run(iso());
db().prepare("UPDATE deployment_servers SET server_deleted_at = ? WHERE server = 'one'").run(iso());
expect(serverSnapshot('one').operations).toEqual([]);
});
it('uses a server index for recent lifecycle history', async () => {
const { db } = await import('../../src/lib/server/db/index.js');
const { serverOperations } = await import('../../src/lib/server/fleet/overview.js');
const prepare = vi.spyOn(db(), 'prepare');
serverOperations('one', now);
const sql = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM server_deployments'))!;
prepare.mockRestore();
const plan = db().prepare(`EXPLAIN QUERY PLAN ${sql}`).all({ server: 'one', since: iso(-86_400_000) }) as { detail: string }[];
const details = plan.map(row => row.detail).join('\n');
expect(details).toMatch(/SEARCH server_deployments USING (?:COVERING )?INDEX idx_server_deployments_server/);
expect(details).not.toMatch(/SCAN server_deployments\b/);
});
});
+73
View File
@@ -145,6 +145,79 @@ describe('game-file import lifecycle', () => {
}); });
}); });
describe('verified host distribution', () => { describe('verified host distribution', () => {
it('reuses published metadata across host reports without reading or parsing a large manifest', () => {
const version = begin();
const manifest = JSON.stringify(Array.from({ length: 20_000 }, (_, i) => ({ ...file, path: `zone/file-${i}.ff` })));
const size = file.size * 20_000;
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=?,manifest=?,manifest_sha256='large-checksum' WHERE id=?")
.run(size, manifest, version.id);
const otherHost = createHost({ name: 'Other reporting host' });
const parse = JSON.parse;
const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => {
if (value === manifest) throw new Error('Cache reports must not deserialize manifests.');
return parse(value, reviver);
});
const preparing = vi.spyOn(state.handle!, 'prepare');
try {
for (const id of [hostId, otherHost.id, hostId]) {
updateHostLibraryCaches(id, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'large-checksum' }]);
expect(getHostLibraryCache(id, version.id)).toMatchObject({ status: 'ready', bytesDownloaded: size, bytesTotal: size });
}
const reads = preparing.mock.calls.map(([sql]) => sql).filter(sql => sql.includes('FROM game_file_versions'));
expect(reads).toHaveLength(1);
expect(state.handle!.prepare(reads[0]).columns().map(column => column.name)).toEqual(['state', 'sizeBytes', 'manifestSha256']);
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'wrong' }]);
expect(getHostLibraryCache(hostId, version.id)?.status).toBe('failed');
expect(() => updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'downloading', bytesDownloaded: size + 1 }]))
.toThrow('Invalid library download progress');
} finally { parsing.mockRestore(); preparing.mockRestore(); }
});
it('does not cache missing or unpublished versions', () => {
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
const version = begin();
state.handle!.prepare("UPDATE game_file_versions SET id='future' WHERE id=?").run(version.id);
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
expect(getHostLibraryCache(hostId, 'future')).toBeNull();
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='checksum' WHERE id='future'").run();
updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
expect(getHostLibraryCache(hostId, 'future')).toMatchObject({ status: 'ready', bytesTotal: 123 });
});
it('does not retain metadata from a rolled-back publication', () => {
const version = begin();
expect(() => state.handle!.transaction(() => {
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='temporary' WHERE id=?").run(version.id);
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
throw new Error('rollback');
})()).toThrow('rollback');
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
expect(getHostLibraryCache(hostId, version.id)).toBeNull();
state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=456,manifest_sha256='published' WHERE id=?").run(version.id);
updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'published' }]);
expect(getHostLibraryCache(hostId, version.id)).toMatchObject({ status: 'ready', bytesTotal: 456 });
});
it('keeps metadata caches separate for each database connection', () => {
const seed = () => {
listLibraries();
legacyLibrary('cache-library', 't6');
legacyVersion('cache-version', 'cache-library', '2025-01-01', '2025-01-01');
};
const report = () => updateHostLibraryCaches(hostId, [{ libraryVersionId: 'cache-version', status: 'ready', manifestSha256: 'checksum' }]);
seed(); report();
const original = state.handle!;
state.handle = new Database(':memory:');
try {
seed();
const preparing = vi.spyOn(state.handle, 'prepare');
try {
report();
expect(preparing.mock.calls.filter(([sql]) => sql.includes('FROM game_file_versions'))).toHaveLength(1);
} finally { preparing.mockRestore(); }
} finally { state.handle.close(); state.handle = original; }
});
it('deduplicates active downloads and blocks readiness without exact manifest verification', async () => { it('deduplicates active downloads and blocks readiness without exact manifest verification', async () => {
const version = await published(); const version = await published();
const operation = queueLibraryDownload(hostId, version.id); const operation = queueLibraryDownload(hostId, version.id);
+17
View File
@@ -68,6 +68,23 @@ function verified(id: string, version: GameLibraryVersion) {
} }
describe('automatic host preparation', () => { describe('automatic host preparation', () => {
it('accepts heartbeats without running controller library cleanup', async () => {
const host = hosts.createHost({ name: 'Heartbeat host' });
const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js');
const cleanup = vi.spyOn(libraries, 'cleanupControllerLibraries').mockRejectedValue(new Error('Cleanup is independent of heartbeats'));
try {
for (let i = 0; i < 3; i++) {
const response = await POST({ request: new Request('https://controller.test/api/agent/heartbeat', {
method: 'POST', headers: { Authorization: `Bearer ${credential}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ inventory: { ...inventory, libraryCaches: [] } })
}) });
expect(response.status).toBe(200);
}
expect(cleanup).not.toHaveBeenCalled();
} finally { cleanup.mockRestore(); }
});
it('keeps re-enabled hosts disabled until a fresh runtime pull and both game libraries are verified', async () => { it('keeps re-enabled hosts disabled until a fresh runtime pull and both game libraries are verified', async () => {
const t5 = await publish('t5'); const t5 = await publish('t5');
const t6 = await publish('t6'); const t6 = await publish('t6');
+146
View File
@@ -0,0 +1,146 @@
import { setImmediate } from 'node:timers/promises';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { HostInventory } from '../../src/lib/hosts.js';
import { newFixture, type Fixture } from '../helpers/fixture.js';
let fixture: Fixture;
const epoch = Date.parse('2026-10-05T12:00:00Z');
const inventory = { hostname: 'request-cache-host', protocolVersion: 1, dockerRunning: true, directoriesWritable: true,
storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }] };
beforeEach(async () => {
vi.resetModules(); vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(epoch);
fixture = await newFixture([]);
});
afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); await fixture.cleanup(); });
async function enrolled() {
const hosts = await import('../../src/lib/server/hosts/index.js');
const context = await import('../../src/lib/server/hosts/request-context.js');
const host = hosts.createHost({ name: 'Request cache host' });
const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
// Finish enrollment's independent availability refresh before counting parses.
await setImmediate();
return { ...hosts, ...context, hostId: host.id, credential };
}
describe('agent credential usage writes', () => {
it('writes on first use and at most once per minute, including after reopening the database', async () => {
const api = await enrolled();
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
db().exec(`CREATE TABLE credential_usage_writes (credential_id TEXT, at TEXT);
CREATE TRIGGER count_credential_usage AFTER UPDATE OF last_used_at ON ec_host_credentials
BEGIN INSERT INTO credential_usage_writes VALUES (NEW.id, NEW.last_used_at); END;`);
const writes = () => db().prepare('SELECT at FROM credential_usage_writes').all();
const authenticate = () => expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
authenticate(); authenticate(); authenticate();
expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }]);
vi.setSystemTime(epoch + 59_999);
authenticate();
closeDb(); authenticate();
expect(writes()).toHaveLength(1);
vi.setSystemTime(epoch + 60_000);
authenticate(); authenticate();
expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }, { at: new Date(epoch + 60_000).toISOString() }]);
expect(api.authenticateCredential('invalid')).toBeNull();
expect(writes()).toHaveLength(2);
});
it('checks revocation immediately and records first use of a replacement credential', async () => {
const api = await enrolled();
const { db } = await import('../../src/lib/server/db/index.js');
await api.withHostRequestContext(() => {
expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
vi.setSystemTime(epoch + 1000);
const replacement = api.rotateCredential(api.hostId);
expect(api.authenticateCredential(api.credential)).toBeNull();
expect(api.authenticateCredential(replacement)?.id).toBe(api.hostId);
expect(db().prepare('SELECT last_used_at FROM ec_host_credentials WHERE revoked_at IS NULL').get())
.toEqual({ last_used_at: new Date(epoch + 1000).toISOString() });
api.revokeCredentials(api.hostId);
expect(api.authenticateCredential(replacement)).toBeNull();
expect(api.getHost(api.hostId)?.credentialActive).toBe(false);
});
});
});
describe('request-scoped host inventory parsing', () => {
it('parses once across authentication, host reads, and operation handling within a request', async () => {
const api = await enrolled();
const serialized = JSON.stringify(inventory);
const parse = vi.spyOn(JSON, 'parse');
await api.withHostRequestContext(async () => {
const authenticated = api.authenticateCredential(api.credential)!;
const operation = api.queueHostOperation(api.hostId, 'instance.inspect', { name: 'server' });
api.pendingHostOperationRows(api.hostId, 'read', false);
api.reportHostOperationStatus(api.hostId, operation.id, { state: 'accepted' });
await Promise.resolve();
for (let i = 0; i < 10; i++) expect(api.getHost(api.hostId)?.inventory).toBe(authenticated.inventory);
expect(api.requireReadyHost(api.hostId).inventory).toBe(authenticated.inventory);
expect(api.listHosts()[0].inventory).toBe(authenticated.inventory);
});
expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(1);
await api.withHostRequestContext(() => api.getHost(api.hostId));
expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(2);
});
it('keeps host metadata fresh and reparses changed inventory, including after rollback', async () => {
const api = await enrolled();
const { db } = await import('../../src/lib/server/db/index.js');
await api.withHostRequestContext(() => {
const original = api.getHost(api.hostId)!;
const updated = api.updateHost(api.hostId, { name: 'Renamed host', mode: 'draining' });
expect(updated).toMatchObject({ name: 'Renamed host', mode: 'draining', readiness: 'draining' });
expect(updated.inventory).toBe(original.inventory);
const heartbeat = api.heartbeat(api.hostId, { ...inventory, storageAvailable: 123 });
expect(heartbeat.inventory).not.toBe(original.inventory);
expect(api.getHost(api.hostId)?.inventory).toBe(heartbeat.inventory);
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
expect(() => db().transaction(() => {
db().prepare('UPDATE ec_hosts SET inventory=? WHERE id=?').run(JSON.stringify({ ...inventory, storageAvailable: 456 }), api.hostId);
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(456);
throw new Error('rollback');
})()).toThrow('rollback');
expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
});
});
it('isolates overlapping requests and does not cache reads outside a request', async () => {
const api = await enrolled();
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
let first!: HostInventory;
const request = api.withHostRequestContext(async () => {
first = api.getHost(api.hostId)!.inventory;
await gate;
expect(api.getHost(api.hostId)!.inventory).toBe(first);
});
try {
await api.withHostRequestContext(async () => {
const second = api.getHost(api.hostId)!.inventory;
expect(second).not.toBe(first);
await Promise.resolve();
expect(api.getHost(api.hostId)!.inventory).toBe(second);
});
expect(api.getHost(api.hostId)!.inventory).not.toBe(api.getHost(api.hostId)!.inventory);
} finally { release(); await request; }
});
it.each([false, true])('releases cached inventory from inherited work after request completion (failed: %s)', async failed => {
const api = await enrolled();
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
let detached!: Promise<void>;
let original!: HostInventory;
const request = api.withHostRequestContext(() => {
original = api.getHost(api.hostId)!.inventory;
detached = gate.then(() => {
const later = api.getHost(api.hostId)!.inventory;
expect(later).not.toBe(original);
expect(api.getHost(api.hostId)!.inventory).not.toBe(later);
});
if (failed) throw new Error('request failed');
});
if (failed) await expect(request).rejects.toThrow('request failed'); else await request;
release(); await detached;
});
});
@@ -70,10 +70,10 @@ describe('provisioning reads with retained lifecycle history', () => {
expect(queries).toHaveLength(1); expect(queries).toHaveLength(1);
const statement = handle.prepare(queries[0]); const statement = handle.prepare(queries[0]);
expect(statement.columns().map(column => column.name)).toEqual(['id', 'server', 'action', 'state', 'stage', 'error', 'started_at']); expect(statement.columns().map(column => column.name)).toEqual(['id', 'server', 'action', 'state', 'stage', 'error', 'started_at']);
const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all(iso(-86_400_000)) as { detail: string }[]; const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all() as { detail: string }[];
const details = plan.map(row => row.detail).join('\n'); const details = plan.map(row => row.detail).join('\n');
expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_state'); expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_state');
expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_finished_at'); expect(queries[0]).not.toContain('finished_at >=');
expect(details).toContain('SEARCH server_deployments USING INTEGER PRIMARY KEY (rowid=?)'); expect(details).toContain('SEARCH server_deployments USING INTEGER PRIMARY KEY (rowid=?)');
expect(details).toContain('SCAN server_deployments USING COVERING INDEX idx_server_deployments_server'); expect(details).toContain('SCAN server_deployments USING COVERING INDEX idx_server_deployments_server');
expect(details).not.toMatch(/SCAN server_deployments(?:\n|$)/); expect(details).not.toMatch(/SCAN server_deployments(?:\n|$)/);
+47
View File
@@ -16,6 +16,53 @@ afterEach(async () => {
}); });
describe('sweep', () => { describe('sweep', () => {
it('emits server state only when state, players, or error changes', async () => {
const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');
const { db } = await import('../../src/lib/server/db/index.js');
const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
await seedServerRegistry();
db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by) VALUES ('changes', 'test', ?, 'ui')")
.run(new Date().toISOString());
db().prepare(`INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state)
VALUES ('changes', 't5_zombie_theater_1', 1, 'pending')`).run();
const events = vi.fn();
const unsubscribe = subscribe(event => { if (event.type === 'server-state') events(event); });
try {
await sweep();
expect(events).toHaveBeenCalledTimes(1);
await sweep();
expect(events).toHaveBeenCalledTimes(1);
// Fresh check timestamps still update the campaign, without an SSE event.
const checkedAt = new Date().toISOString();
db().prepare('UPDATE server_deployments SET checked_at = ?').run(checkedAt);
await sweep();
expect(events).toHaveBeenCalledTimes(1);
expect(db().prepare("SELECT checked_at, attempts FROM deployment_servers WHERE deployment_id = 'changes'").get())
.toEqual({ checked_at: checkedAt, attempts: 1 });
db().prepare('UPDATE server_deployments SET players = 2').run();
await sweep();
expect(events).toHaveBeenCalledTimes(2);
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'pending', players: 2 }));
db().prepare("UPDATE server_deployments SET error = 'Waiting for players'").run();
await sweep();
expect(events).toHaveBeenCalledTimes(3);
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ error: 'Waiting for players' }));
db().prepare("UPDATE server_deployments SET state = 'failed'").run();
await sweep();
expect(events).toHaveBeenCalledTimes(4);
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'failed' }));
await sweep();
expect(events).toHaveBeenCalledTimes(4);
db().prepare("UPDATE server_deployments SET state = 'done'").run();
await sweep();
expect(events).toHaveBeenCalledTimes(5);
expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'done', error: null }));
} finally { unsubscribe(); }
});
it('queues pending rows in Provisioning even when integrations are unset', async () => { it('queues pending rows in Provisioning even when integrations are unset', async () => {
const { sweep } = await import('../../src/lib/server/pipeline/sweep.js'); const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js'); const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');
+39
View File
@@ -90,6 +90,45 @@ it('timestamps initial telemetry and subsequent status and telemetry events with
} }
}); });
it('serializes each live server event once across connections and releases cancelled subscribers', async () => {
const { GET } = await import('../../src/routes/api/events/servers/+server.js');
const { emit } = await import('../../src/lib/server/pipeline/bus.js');
const readers = await Promise.all([GET(), GET()].map(async response => (await response).body!.getReader()));
const stringify = vi.spyOn(JSON, 'stringify');
try {
for (const reader of readers) {
await reader.read(); // Initial comment and per-connection snapshots.
await reader.read();
await reader.read();
}
for (const event of [
{ type: 'server-status' as const, delta: true, entries: [] },
{ type: 'server-telemetry' as const, entries: [{ ...payload, receivedAt: new Date().toISOString() }] }
]) {
stringify.mockClear();
emit(event);
expect(stringify).toHaveBeenCalledTimes(1);
const [first, second] = await Promise.all(readers.map(reader => reader.read()));
expect(first.value).toBe(second.value);
const frame = new TextDecoder().decode(first.value);
expect(frame.startsWith(`event: ${event.type}\ndata: `)).toBe(true);
expect(JSON.parse(frame.split('\n')[1].slice(6))).toEqual({ ...event, serverNow: expect.any(Number) });
}
await readers[0].cancel();
stringify.mockClear();
emit({ type: 'server-status', entries: [] });
expect(stringify).toHaveBeenCalledTimes(1);
expect((await readers[1].read()).done).toBe(false);
await readers[1].cancel();
stringify.mockClear();
emit({ type: 'server-telemetry', entries: [] });
expect(stringify).not.toHaveBeenCalled();
} finally {
stringify.mockRestore();
await Promise.all(readers.map(reader => reader.cancel()));
}
});
it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => { it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => {
const payload = gameEvent(event, 17); const payload = gameEvent(event, 17);
const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
+154 -7
View File
@@ -8,9 +8,16 @@ import { gameEvent } from '../helpers/telemetry.js';
vi.mock('$app/environment', () => ({ browser: true })); vi.mock('$app/environment', () => ({ browser: true }));
const epoch = Date.parse('2026-09-19T12:00:00Z'); const epoch = Date.parse('2026-09-19T12:00:00Z');
let source: EventTarget; let source: EventTarget & { close: () => void };
let stop: (() => void) | undefined; let stop: (() => void) | undefined;
let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object }; let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object };
let frames: Map<number, FrameRequestCallback>;
function flushFrame() {
const callbacks = [...frames.values()];
frames.clear();
for (const callback of callbacks) callback(performance.now());
}
function server(timestamp = epoch, changes: Partial<FleetServer> = {}): FleetServer { function server(timestamp = epoch, changes: Partial<FleetServer> = {}): FleetServer {
return { return {
@@ -35,12 +42,20 @@ function status(timestamp: number, changes: Partial<ServerStatus> = {}): ServerS
function emit(serverNow: number, entries: ServerStatus[]) { function emit(serverNow: number, entries: ServerStatus[]) {
source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) })); source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) }));
flushFrame();
} }
beforeEach(() => { beforeEach(() => {
vi.resetModules(); vi.resetModules();
vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] }); vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] });
vi.setSystemTime(epoch); vi.setSystemTime(epoch);
frames = new Map();
let frameId = 0;
vi.stubGlobal('requestAnimationFrame', vi.fn((callback: FrameRequestCallback) => {
frames.set(++frameId, callback);
return frameId;
}));
vi.stubGlobal('cancelAnimationFrame', vi.fn((id: number) => { frames.delete(id); }));
snapshot = { serverNow: epoch, servers: [server()], integrations: {} }; snapshot = { serverNow: epoch, servers: [server()], integrations: {} };
vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify( vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify(
url === '/api/servers' ? snapshot : { operations: [] } url === '/api/servers' ? snapshot : { operations: [] }
@@ -66,6 +81,130 @@ async function setup() {
return { ...clock, ...fleet }; return { ...clock, ...fleet };
} }
function queueEvent(type: string, data: object) {
source.dispatchEvent(new MessageEvent(type, { data: JSON.stringify(data) }));
}
describe('animation-frame stream batching', () => {
it('publishes each changed store once for a burst and preserves lifecycle evidence through heartbeats', async () => {
const stores = await setup();
emit(epoch, [status(epoch), status(epoch, { server: 'two' })]);
const before = get(stores.statusByServer);
await vi.advanceTimersByTimeAsync(1000);
const watched = [stores.statusByServer, stores.statusReceivedAt, stores.telemetryByServer, stores.matchEventsByServer, stores.uiNow];
const changes = watched.map(() => vi.fn());
const unsubscribe = watched.map((store, index) => store.subscribe(changes[index]));
changes.forEach(change => change.mockClear());
vi.mocked(requestAnimationFrame).mockClear();
try {
for (let i = 0; i < 100; i++) {
queueEvent('server-status', { delta: true, serverNow: epoch + 1000 + i,
entries: [status(epoch + 1000, { players: i })] });
queueEvent('server-telemetry', { serverNow: epoch + 1000 + i,
entries: [{ ...gameEvent(i === 0 ? 'game_start' : 'heartbeat', i, 1), receivedAt: new Date(epoch + 1000 + i).toISOString() }] });
}
expect(requestAnimationFrame).toHaveBeenCalledTimes(1);
changes.forEach(change => expect(change).not.toHaveBeenCalled());
expect(get(stores.statusByServer)).toBe(before);
expect(before.get('t5_zombie_theater_1')?.players).toBe(0);
flushFrame();
changes.forEach(change => expect(change).toHaveBeenCalledTimes(1));
expect(get(stores.statusByServer).size).toBe(2);
expect(get(stores.statusByServer).get('t5_zombie_theater_1')?.players).toBe(99);
expect(get(stores.telemetryByServer).get('t5_zombie_theater_1')).toMatchObject({ event: 'heartbeat', round: 99 });
expect(get(stores.matchEventsByServer).get('t5_zombie_theater_1')?.event).toBe('game_start');
queueEvent('server-telemetry', { entries: [{ ...gameEvent('heartbeat', 100, 1), receivedAt: new Date(epoch + 1100).toISOString() }] });
flushFrame();
expect(requestAnimationFrame).toHaveBeenCalledTimes(2);
expect(changes[2]).toHaveBeenCalledTimes(2);
expect(changes[3]).toHaveBeenCalledTimes(1);
} finally { unsubscribe.forEach(unsubscribe => unsubscribe()); }
});
it('applies snapshots and subsequent deltas in order, including empty snapshots', async () => {
const { statusByServer, telemetryByServer, matchEventsByServer } = await setup();
const telemetry = (event: GameEvent['event'], name: string) => ({ ...gameEvent(event, 1, 1, name), receivedAt: new Date(epoch).toISOString() });
emit(epoch, [status(epoch, { server: 'old' })]);
queueEvent('server-telemetry', { entries: [telemetry('game_start', 'old')] });
flushFrame();
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'discarded' })] });
queueEvent('server-status', { entries: [status(epoch, { server: 'two' })] });
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'three' })] });
queueEvent('server-telemetry', { entries: [telemetry('game_end', 'discarded')] });
queueEvent('server-telemetry', { snapshot: true, entries: [telemetry('heartbeat', 'two')] });
queueEvent('server-telemetry', { entries: [telemetry('game_end', 'three')] });
queueEvent('server-telemetry', { entries: [telemetry('heartbeat', 'three')] });
flushFrame();
expect([...get(statusByServer).keys()]).toEqual(['two', 'three']);
expect([...get(telemetryByServer).keys()]).toEqual(['two', 'three']);
expect([...get(matchEventsByServer).keys()]).toEqual(['three']);
expect(get(matchEventsByServer).get('three')?.event).toBe('game_end');
expect(get(telemetryByServer).get('three')?.event).toBe('heartbeat');
queueEvent('server-status', { entries: [] });
queueEvent('server-telemetry', { snapshot: true, entries: [] });
flushFrame();
expect(get(statusByServer).size).toBe(0);
expect(get(telemetryByServer).size).toBe(0);
expect(get(matchEventsByServer).size).toBe(0);
});
it('retains event arrival times so buffered status cannot overwrite a newer HTTP snapshot', async () => {
const { fleetServers, refreshFleet, statusByServer, statusReceivedAt } = await setup();
await vi.advanceTimersByTimeAsync(1000);
queueEvent('server-status', { delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 1 })] });
await vi.advanceTimersByTimeAsync(1000);
snapshot = { ...snapshot, serverNow: epoch + 2000, servers: [server(epoch + 2000, { last_players: 4 })] };
await refreshFleet();
queueEvent('server-status', { delta: true, serverNow: epoch + 2000, entries: [status(epoch + 2000, { server: 'two' })] });
flushFrame();
expect(get(statusByServer).get('t5_zombie_theater_1')?.receivedAt).toBe(epoch + 1000);
expect(get(statusReceivedAt)).toBe(epoch + 2000);
expect(get(fleetServers)[0].last_players).toBe(4);
});
it('accounts for frame delay without rewinding the server clock', async () => {
const { uiNow, statusByServer } = await setup();
vi.stubGlobal('document', { visibilityState: 'hidden' });
queueEvent('server-status', { serverNow: epoch + 5000, entries: [status(epoch + 5000)] });
await vi.advanceTimersByTimeAsync(2000);
queueEvent('server-telemetry', { serverNow: epoch + 4000, entries: [] });
await vi.advanceTimersByTimeAsync(1000);
expect(get(uiNow)).toBe(epoch);
expect(get(statusByServer).size).toBe(0);
expect(frames.size).toBe(1);
flushFrame();
expect(get(uiNow)).toBe(epoch + 8000);
expect(get(statusByServer).size).toBe(1);
});
it('cancels queued frames and discards buffered events when stopping and restarting', async () => {
const { statusByServer, telemetryByServer, startServerStream } = await setup();
const oldSource = source;
const close = vi.spyOn(oldSource, 'close');
queueEvent('server-status', { entries: [status(epoch)] });
queueEvent('server-telemetry', { entries: [{ ...gameEvent('game_start'), receivedAt: new Date(epoch).toISOString() }] });
const callback = [...frames.values()][0];
const oldStop = stop!;
oldStop();
expect(close).toHaveBeenCalledTimes(1);
expect(cancelAnimationFrame).toHaveBeenCalledTimes(1);
expect(frames.size).toBe(0);
callback(0);
expect(get(statusByServer).size).toBe(0);
expect(get(telemetryByServer).size).toBe(0);
stop = startServerStream();
oldStop();
oldSource.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ entries: [status(epoch)] }) }));
queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'new' })] });
flushFrame();
expect([...get(statusByServer).keys()]).toEqual(['new']);
expect(get(telemetryByServer).size).toBe(0);
});
});
describe('Fleet report clock', () => { describe('Fleet report clock', () => {
it('accepts newer navigation snapshots while keeping stream evidence newer than the snapshot', async () => { it('accepts newer navigation snapshots while keeping stream evidence newer than the snapshot', async () => {
const { fleetServers, hydrateFleet } = await setup(); const { fleetServers, hydrateFleet } = await setup();
@@ -86,6 +225,7 @@ describe('Fleet report clock', () => {
source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({
delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 2 })] delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 2 })]
}) })); }) }));
flushFrame();
expect(get(statusByServer).size).toBe(2); expect(get(statusByServer).size).toBe(2);
expect(get(fleetServers).find(server => server.name === 't5_zombie_theater_1')?.last_players).toBe(2); expect(get(fleetServers).find(server => server.name === 't5_zombie_theater_1')?.last_players).toBe(2);
const changes = vi.fn(); const changes = vi.fn();
@@ -103,6 +243,7 @@ describe('Fleet report clock', () => {
source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + seconds * 1000, source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + seconds * 1000,
entries: [{ ...gameEvent(event, 2, players), receivedAt: at(seconds), entries: [{ ...gameEvent(event, 2, players), receivedAt: at(seconds),
timing: { matchId: 1, players: {}, server: timing } }] }) })); timing: { matchId: 1, players: {}, server: timing } }] }) }));
flushFrame();
return get(fleetServers)[0]; return get(fleetServers)[0];
}; };
const active = { activeSince: at(1), inactiveSince: null, gameStartedAt: at(1), gameEndedAt: null }; const active = { activeSince: at(1), inactiveSince: null, gameStartedAt: at(1), gameEndedAt: null };
@@ -132,15 +273,17 @@ describe('Fleet report clock', () => {
entries: [{ ...gameEvent('heartbeat', 2, 1), receivedAt: new Date(epoch).toISOString(), entries: [{ ...gameEvent('heartbeat', 2, 1), receivedAt: new Date(epoch).toISOString(),
timing: { matchId: 1, players: {}, server: { activeSince: startedAt, inactiveSince: null, timing: { matchId: 1, players: {}, server: { activeSince: startedAt, inactiveSince: null,
gameStartedAt: startedAt, gameEndedAt: null } } }] }) })); gameStartedAt: startedAt, gameEndedAt: null } } }] }) }));
flushFrame();
expect(get(fleetServers)[0]).toMatchObject({ active_since: startedAt, game_started_at: startedAt }); expect(get(fleetServers)[0]).toMatchObject({ active_since: startedAt, game_started_at: startedAt });
}); });
it('updates count and roster immediately from telemetry and expires them together', async () => { it('updates count and roster on the next frame and expires them together', async () => {
snapshot.servers = [server(epoch, { container_state: 'running', container_checked_at: new Date(epoch).toISOString(), last_players: 4 })]; snapshot.servers = [server(epoch, { container_state: 'running', container_checked_at: new Date(epoch).toISOString(), last_players: 4 })];
const { fleetServers, uiNow } = await setup(); const { fleetServers, uiNow } = await setup();
await vi.advanceTimersByTimeAsync(1000); await vi.advanceTimersByTimeAsync(1000);
source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + 1000, source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + 1000,
entries: [{ ...gameEvent('heartbeat', 15, 3), receivedAt: new Date(epoch + 1000).toISOString() }] }) })); entries: [{ ...gameEvent('heartbeat', 15, 3), receivedAt: new Date(epoch + 1000).toISOString() }] }) }));
flushFrame();
expect(get(fleetServers)[0]).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry' }); expect(get(fleetServers)[0]).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry' });
expect(get(fleetServers)[0].roster).toHaveLength(3); expect(get(fleetServers)[0].roster).toHaveLength(3);
expect(matchesFleet(get(fleetServers)[0], 'Player 3', 'occupied', '', [], [], get(uiNow))).toBe(true); expect(matchesFleet(get(fleetServers)[0], 'Player 3', 'occupied', '', [], [], get(uiNow))).toBe(true);
@@ -149,12 +292,15 @@ describe('Fleet report clock', () => {
expect(matchesFleet(get(fleetServers)[0], '', 'occupied', '', [], [], get(uiNow))).toBe(false); expect(matchesFleet(get(fleetServers)[0], '', 'occupied', '', [], [], get(uiNow))).toBe(false);
}); });
it('applies game lifecycle events immediately and clears them after an empty report', async () => { it('applies game lifecycle events on the next frame and clears them after an empty report', async () => {
const { fleetServers } = await setup(); const { fleetServers } = await setup();
const report = (event: string, time: number) => source.dispatchEvent(new MessageEvent('server-telemetry', { const report = (event: string, time: number) => {
data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event, source.dispatchEvent(new MessageEvent('server-telemetry', {
receivedAt: new Date(time).toISOString(), players: [] }] }) data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event,
})); receivedAt: new Date(time).toISOString(), players: [] }] })
}));
flushFrame();
};
emit(epoch, [status(epoch, { players: 1 })]); emit(epoch, [status(epoch, { players: 1 })]);
report('game_end', epoch + 1000); report('game_end', epoch + 1000);
expect(get(fleetServers)[0].game_ended_at).toBe(new Date(epoch + 1000).toISOString()); expect(get(fleetServers)[0].game_ended_at).toBe(new Date(epoch + 1000).toISOString());
@@ -249,6 +395,7 @@ describe('Fleet report clock', () => {
source.dispatchEvent(new MessageEvent('server-telemetry', { source.dispatchEvent(new MessageEvent('server-telemetry', {
data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true }) data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true })
})); }));
flushFrame();
expect(get(uiNow)).toBe(epoch); expect(get(uiNow)).toBe(epoch);
await vi.advanceTimersByTimeAsync(1000); await vi.advanceTimersByTimeAsync(1000);
expect(get(uiNow)).toBe(epoch + 1000); expect(get(uiNow)).toBe(epoch + 1000);
+24
View File
@@ -31,6 +31,30 @@ async function setup() {
} }
describe('independent provisioning scheduling', () => { describe('independent provisioning scheduling', () => {
it('runs controller library cleanup on worker ticks without overlapping a slow cleanup', async () => {
const s = await setup();
const { cleanupControllerLibraries } = await import('../../src/lib/server/game-files/index.js');
const cleanup = deferred();
vi.mocked(cleanupControllerLibraries).mockImplementationOnce(() => cleanup.promise);
vi.useFakeTimers();
try {
s.startWorker();
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(30_000);
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
expect(s.process.mock.calls.length).toBeGreaterThan(1);
cleanup.resolve();
await vi.advanceTimersByTimeAsync(0);
await vi.advanceTimersByTimeAsync(15_000);
expect(cleanupControllerLibraries).toHaveBeenCalledTimes(2);
} finally {
s.stopWorker(); cleanup.resolve(); s.background.resolve();
await vi.advanceTimersByTimeAsync(0);
vi.useRealTimers();
await vi.waitFor(() => expect(s.workerActivity()).toEqual([]));
}
});
it('admits and checks new work while campaign, beta and maintenance observers are still waiting', async () => { it('admits and checks new work while campaign, beta and maintenance observers are still waiting', async () => {
const s = await setup(); const s = await setup();
try { try {